Add large-payload blob auto-purge (worker/SDK side)

Microsoft.DurableTask.sln

@@ -1,4 +1,4 @@
-
+
 Microsoft Visual Studio Solution File, Format Version 12.00
 # Visual Studio Version 17
 VisualStudioVersion = 17.3.32901.215
@@ -145,6 +145,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "AzureManaged", "AzureManage
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Grpc", "Grpc", "{3B8F957E-7773-4C0C-ACD7-91A1591D9312}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "AzureBlobPayloads.Tests", "test\AzureBlobPayloads.Tests\AzureBlobPayloads.Tests.csproj", "{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -839,6 +841,18 @@ Global
 		{C1995163-1DCE-405D-BE82-8B4B2584893E}.Release|x64.Build.0 = Release|Any CPU
 		{C1995163-1DCE-405D-BE82-8B4B2584893E}.Release|x86.ActiveCfg = Release|Any CPU
 		{C1995163-1DCE-405D-BE82-8B4B2584893E}.Release|x86.Build.0 = Release|Any CPU
+		{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Debug|x64.Build.0 = Debug|Any CPU
+		{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Debug|x86.Build.0 = Debug|Any CPU
+		{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Release|Any CPU.Build.0 = Release|Any CPU
+		{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Release|x64.ActiveCfg = Release|Any CPU
+		{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Release|x64.Build.0 = Release|Any CPU
+		{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Release|x86.ActiveCfg = Release|Any CPU
+		{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Release|x86.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -911,6 +925,7 @@ Global
 		{C1995163-1DCE-405D-BE82-8B4B2584893E} = {9686B8F9-2644-6C9B-E567-55B0471E4584}
 		{53193780-CD18-2643-6953-C26F59EAEDF5} = {5B448FF6-EC42-491D-A22E-1DC8B618E6D5}
 		{3B8F957E-7773-4C0C-ACD7-91A1591D9312} = {5B448FF6-EC42-491D-A22E-1DC8B618E6D5}
+		{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5} = {E5637F81-2FB9-4CD7-900D-455363B142A7}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {AB41CB55-35EA-4986-A522-387AB3402E71}

src/Extensions/AzureBlobPayloads/AutoPurge/BlobPayloadAutoPurgeService.cs

@@ -0,0 +1,233 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using Grpc.Core;
+using Microsoft.DurableTask.Worker.Grpc;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using P = Microsoft.DurableTask.Protobuf;
+
+namespace Microsoft.DurableTask;
+
+/// <summary>
+/// Background service that drains blob-externalized payload tombstones streamed by the Durable Task
+/// backend and deletes the corresponding blobs from customer storage. The backend cannot delete these
+/// blobs itself because it has no storage credentials; it soft-deletes the payload row, pushes the token
+/// to this worker, and hard-deletes the row only once the worker acknowledges the blob has been removed.
+/// </summary>
+/// <remarks>
+/// <para>
+/// On the open-source worker this service is registered automatically by <c>UseExternalizedPayloads</c>
+/// and reuses that worker's gRPC connection. It can also be constructed directly from a
+/// <see cref="CallInvoker"/> (or gRPC channel) and a <see cref="PayloadStore"/> so hosts that enable large
+/// payloads through a different code path (for example the Durable Task Scheduler "Azure Managed" worker
+/// SDK) can reuse it without going through <c>UseExternalizedPayloads</c>. Because it derives from
+/// <see cref="BackgroundService"/> such hosts can start and stop it directly via
+/// <see cref="IHostedService.StartAsync"/> / <see cref="IHostedService.StopAsync"/>.
+/// </para>
+/// <para>
+/// It opens a single bidirectional gRPC stream and blocks awaiting server pushes when idle (a push model,
+/// not a polling loop) for the worker process lifetime. Deletes are idempotent, so running multiple worker
+/// replicas concurrently is safe.
+/// </para>
+/// </remarks>
+public sealed class BlobPayloadAutoPurgeService : BackgroundService
+{
+    static readonly TimeSpan ReconnectBackoffBase = TimeSpan.FromSeconds(1);
+    static readonly TimeSpan ReconnectBackoffCap = TimeSpan.FromSeconds(30);
+
+    readonly PayloadStore store;
+    readonly ILogger logger;
+    readonly Random jitter = new();
+
+    // Exactly one connection strategy is used. When a CallInvoker is supplied directly (cross-repo reuse) it
+    // is used as-is; otherwise the OSS DI path resolves it lazily from the named worker options at start,
+    // which keeps host startup resilient if the worker connection is misconfigured.
+    readonly CallInvoker? callInvoker;
+    readonly IOptionsMonitor<GrpcDurableTaskWorkerOptions>? workerOptions;
+    readonly string? builderName;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="BlobPayloadAutoPurgeService"/> class that drains the
+    /// purge stream over the supplied gRPC call invoker. Use this constructor to reuse the service from a
+    /// host that does not enable large payloads through <c>UseExternalizedPayloads</c>.
+    /// </summary>
+    /// <param name="callInvoker">The gRPC call invoker connected to the Durable Task backend.</param>
+    /// <param name="store">The payload store used to delete blobs.</param>
+    /// <param name="logger">The logger, or <c>null</c> to disable logging.</param>
+    public BlobPayloadAutoPurgeService(
+        CallInvoker callInvoker,
+        PayloadStore store,
+        ILogger? logger = null)
+        : this(store, logger)
+    {
+        this.callInvoker = Check.NotNull(callInvoker);
+    }
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="BlobPayloadAutoPurgeService"/> class that drains the
+    /// purge stream over the supplied gRPC channel. Use this constructor to reuse the service from a host
+    /// that does not enable large payloads through <c>UseExternalizedPayloads</c>.
+    /// </summary>
+    /// <param name="channel">The gRPC channel connected to the Durable Task backend.</param>
+    /// <param name="store">The payload store used to delete blobs.</param>
+    /// <param name="logger">The logger, or <c>null</c> to disable logging.</param>
+    public BlobPayloadAutoPurgeService(
+        ChannelBase channel,
+        PayloadStore store,
+        ILogger? logger = null)
+        : this(Check.NotNull(channel).CreateCallInvoker(), store, logger)
+    {
+    }
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="BlobPayloadAutoPurgeService"/> class that resolves its
+    /// gRPC connection from the named worker options. Used by the <c>UseExternalizedPayloads</c> DI path.
+    /// </summary>
+    /// <param name="store">The payload store used to delete blobs.</param>
+    /// <param name="workerOptions">Monitor for the named gRPC worker options that carry the DTS connection.</param>
+    /// <param name="builderName">The Durable Task worker builder name whose options this service uses.</param>
+    /// <param name="logger">The logger.</param>
+    internal BlobPayloadAutoPurgeService(
+        PayloadStore store,
+        IOptionsMonitor<GrpcDurableTaskWorkerOptions> workerOptions,
+        string builderName,
+        ILogger<BlobPayloadAutoPurgeService> logger)
+        : this(store, logger)
+    {
+        this.workerOptions = Check.NotNull(workerOptions);
+        this.builderName = Check.NotNull(builderName);
+    }
+
+    BlobPayloadAutoPurgeService(PayloadStore store, ILogger? logger)
+    {
+        this.store = Check.NotNull(store);
+        this.logger = logger ?? NullLogger.Instance;
+    }
+
+    /// <inheritdoc/>
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        CallInvoker? callInvoker = this.ResolveCallInvoker();
+        if (callInvoker is null)
+        {
+            // ResolveCallInvoker already logged why the purge stream cannot run.
+            return;
+        }
+
+        P.TaskHubSidecarService.TaskHubSidecarServiceClient client = new(callInvoker);
+        TimeSpan backoff = ReconnectBackoffBase;
+
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            try
+            {
+                await this.DrainPurgeStreamAsync(client, stoppingToken);
+
+                // A clean server-side completion just means "nothing more to purge right now"; reconnect promptly.
+                backoff = ReconnectBackoffBase;
+            }
+            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            {
+                break;
+            }
+            catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException)
+            {
+                Logs.AutoPurgeStreamError(this.logger, ex);
+                await this.DelayWithJitterAsync(backoff, stoppingToken);
+                backoff = NextBackoff(backoff);
+            }
+        }
+    }
+
+    static TimeSpan NextBackoff(TimeSpan current)
+    {
+        if (current >= ReconnectBackoffCap)
+        {
+            return ReconnectBackoffCap;
+        }
+
+        long doubledTicks = current.Ticks * 2;
+        return doubledTicks >= ReconnectBackoffCap.Ticks ? ReconnectBackoffCap : TimeSpan.FromTicks(doubledTicks);
+    }
+
+    CallInvoker? ResolveCallInvoker()
+    {
+        // Cross-repo reuse path: the caller supplied a ready-to-use connection.
+        if (this.callInvoker is not null)
+        {
+            return this.callInvoker;
+        }
+
+        // OSS DI path: reuse the same connection the worker uses to reach DTS. After UseExternalizedPayloads
+        // runs, the named options are guaranteed to expose a CallInvoker (a Channel, if supplied, is folded
+        // into one during PostConfigure).
+        CallInvoker? resolved;
+        try
+        {
+            resolved = this.workerOptions?.Get(this.builderName!).CallInvoker;
+        }
+        catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException)
+        {
+            Logs.AutoPurgeConfigurationError(this.logger, ex);
+            return null;
+        }
+
+        if (resolved is null)
+        {
+            Logs.AutoPurgeNoCallInvoker(this.logger);
+        }
+
+        return resolved;
+    }
+
+    async Task DrainPurgeStreamAsync(
+        P.TaskHubSidecarService.TaskHubSidecarServiceClient client, CancellationToken stoppingToken)
+    {
+        using AsyncDuplexStreamingCall<P.PayloadPurgeAck, P.TombstonedPayload> call =
+            client.PurgeExternalPayloads(cancellationToken: stoppingToken);
+
+        Logs.AutoPurgeStreamOpened(this.logger);
+
+        while (await call.ResponseStream.MoveNext(stoppingToken))
+        {
+            P.TombstonedPayload tombstone = call.ResponseStream.Current;
+            try
+            {
+                await this.store.DeleteAsync(tombstone.Token, stoppingToken);
+            }
+            catch (Exception ex) when (
+                ex is not OperationCanceledException and not OutOfMemoryException and not StackOverflowException)
+            {
+                // Leave the payload soft-deleted so the backend can re-push it later; skip the ack.
+                Logs.AutoPurgeDeleteFailed(this.logger, ex, tombstone.Token);
+                continue;
+            }
+
+            // The ack write inherits cancellation from the duplex call (opened with stoppingToken); the
+            // WriteAsync(message, CancellationToken) overload is unavailable on netstandard2.0.
+#pragma warning disable CA2016 // Forward the CancellationToken parameter to methods that take one
+            await call.RequestStream.WriteAsync(new P.PayloadPurgeAck
+            {
+                PartitionId = tombstone.PartitionId,
+                InstanceKey = tombstone.InstanceKey,
+                PayloadId = tombstone.PayloadId,
+            });
+#pragma warning restore CA2016
+        }
+
+        await call.RequestStream.CompleteAsync();
+    }
+
+    async Task DelayWithJitterAsync(TimeSpan maxDelay, CancellationToken cancellationToken)
+    {
+        // Full jitter in [0, maxDelay) spreads reconnect attempts across replicas.
+        TimeSpan delay = TimeSpan.FromTicks((long)(maxDelay.Ticks * this.jitter.NextDouble()));
+        if (delay > TimeSpan.Zero)
+        {
+            await Task.Delay(delay, cancellationToken);
+        }
+    }
+}

src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs

@@ -0,0 +1,42 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using Microsoft.Extensions.Logging;
+
+namespace Microsoft.DurableTask;
+
+/// <summary>
+/// Log messages for the Azure Blob externalized-payload auto-purge background service.
+/// </summary>
+static partial class Logs
+{
+    [LoggerMessage(
+        EventId = 800,
+        Level = LogLevel.Debug,
+        Message = "Blob payload auto-purge stream opened; awaiting tombstoned payloads from the backend.")]
+    public static partial void AutoPurgeStreamOpened(ILogger logger);
+
+    [LoggerMessage(
+        EventId = 801,
+        Level = LogLevel.Warning,
+        Message = "Blob payload auto-purge stream failed; reconnecting after backoff.")]
+    public static partial void AutoPurgeStreamError(ILogger logger, Exception exception);
+
+    [LoggerMessage(
+        EventId = 802,
+        Level = LogLevel.Warning,
+        Message = "Failed to delete externalized payload blob for token '{Token}'; leaving it for the backend to re-push.")]
+    public static partial void AutoPurgeDeleteFailed(ILogger logger, Exception exception, string token);
+
+    [LoggerMessage(
+        EventId = 803,
+        Level = LogLevel.Warning,
+        Message = "Blob payload auto-purge could not resolve a gRPC call invoker; the purge stream will not run.")]
+    public static partial void AutoPurgeNoCallInvoker(ILogger logger);
+
+    [LoggerMessage(
+        EventId = 804,
+        Level = LogLevel.Error,
+        Message = "Blob payload auto-purge failed to read the worker gRPC options; the purge stream will not run.")]
+    public static partial void AutoPurgeConfigurationError(ILogger logger, Exception exception);
+}

src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs

@@ -7,6 +7,8 @@
 using Microsoft.DurableTask.Worker;
 using Microsoft.DurableTask.Worker.Grpc;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using P = Microsoft.DurableTask.Protobuf;
 
@@ -82,6 +84,20 @@ static IDurableTaskWorkerBuilder UseExternalizedPayloadsCore(IDurableTaskWorkerB
                 opt.Capabilities.Add(P.WorkerCapability.LargePayloads);
             });
 
+        // Drain blob-externalized payload tombstones the backend streams for deletion. Gated on this enable
+        // path: if UseExternalizedPayloads is never called, the service is never registered and never runs.
+        builder.Services.AddSingleton<IHostedService>(
+            sp => CreateAutoPurgeService(sp, builder.Name));
+
         return builder;
     }
+
+    static BlobPayloadAutoPurgeService CreateAutoPurgeService(IServiceProvider services, string builderName)
+    {
+        return new BlobPayloadAutoPurgeService(
+            services.GetRequiredService<PayloadStore>(),
+            services.GetRequiredService<IOptionsMonitor<GrpcDurableTaskWorkerOptions>>(),
+            builderName,
+            services.GetRequiredService<ILogger<BlobPayloadAutoPurgeService>>());
+    }
 }

src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs

@@ -67,6 +67,18 @@ public BlobPayloadStore(LargePayloadStorageOptions options)
         this.containerClient = serviceClient.GetBlobContainerClient(options.ContainerName);
     }
 
+    /// <summary>
+    /// Initializes a new instance of the <see cref="BlobPayloadStore"/> class using a caller-supplied
+    /// container client. Intended for unit testing so a mocked <see cref="BlobContainerClient"/> can be injected.
+    /// </summary>
+    /// <param name="containerClient">The blob container client to use.</param>
+    /// <param name="options">The options for the blob payload store.</param>
+    internal BlobPayloadStore(BlobContainerClient containerClient, LargePayloadStorageOptions options)
+    {
+        this.containerClient = containerClient ?? throw new ArgumentNullException(nameof(containerClient));
+        this.options = options ?? throw new ArgumentNullException(nameof(options));
+    }
+
     /// <inheritdoc/>
     public override async Task<string> UploadAsync(string payLoad, CancellationToken cancellationToken)
     {
@@ -145,6 +157,25 @@ public override async Task<string> DownloadAsync(string token, CancellationToken
         }
     }
 
+    /// <inheritdoc/>
+    public override async Task DeleteAsync(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);
+
+        // Idempotent by design: DeleteIfExistsAsync returns false (rather than throwing) when the blob is
+        // already gone, so re-delivered tombstones and concurrent purges from multiple worker replicas are safe.
+        await blob.DeleteIfExistsAsync(
+            DeleteSnapshotsOption.IncludeSnapshots,
+            conditions: null,
+            cancellationToken);
+    }
+
     /// <inheritdoc/>
     public override bool IsKnownPayloadToken(string value)
     {