Add large-payload blob auto-purge (opt-in singleton job, 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/Client/Core/DurableTaskClient.cs

@@ -549,6 +549,30 @@ public virtual Task<Page<string>> ListInstanceIdsAsync(
             $"{this.GetType()} does not support listing orchestration instance IDs filtered by completed time.");
     }
 
+    /// <summary>
+    /// Gets a batch of tombstoned (soft-deleted) externalized payloads whose backing blobs should be deleted
+    /// by a credentialed caller before the backend hard-deletes the rows.
+    /// </summary>
+    /// <param name="limit">The maximum number of tombstoned payloads to request.</param>
+    /// <param name="cancellation">The cancellation token.</param>
+    /// <returns>The batch of tombstoned payloads whose blobs should be deleted.</returns>
+    /// <exception cref="NotSupportedException">Thrown if this implementation does not support the operation.</exception>
+    public virtual Task<List<TombstonedPayload>> GetTombstonedPayloadsAsync(
+        int limit, CancellationToken cancellation = default)
+        => throw new NotSupportedException($"{this.GetType()} does not support retrieving tombstoned payloads.");
+
+    /// <summary>
+    /// Acknowledges tombstoned payloads whose backing blobs have been deleted so the backend can hard-delete
+    /// the corresponding rows.
+    /// </summary>
+    /// <param name="acks">The payloads whose blobs have been deleted.</param>
+    /// <param name="cancellation">The cancellation token.</param>
+    /// <returns>A task that completes when the acknowledgement has been sent.</returns>
+    /// <exception cref="NotSupportedException">Thrown if this implementation does not support the operation.</exception>
+    public virtual Task AckPurgedPayloadsAsync(
+        IEnumerable<PayloadPurgeAck> acks, CancellationToken cancellation = default)
+        => throw new NotSupportedException($"{this.GetType()} does not support acknowledging purged payloads.");
+
     // TODO: Create task hub
 
     // TODO: Delete task hub

src/Client/Core/PayloadPurgeAck.cs

@@ -0,0 +1,13 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace Microsoft.DurableTask.Client;
+
+/// <summary>
+/// Serializable acknowledgement that the worker has deleted the blob for a tombstoned payload, so the
+/// backend can hard-delete the soft-deleted row. Mirrors the <c>PayloadPurgeAck</c> protobuf message.
+/// </summary>
+/// <param name="PartitionId">The backend partition that owns the payload row.</param>
+/// <param name="InstanceKey">The orchestration instance key the payload belongs to.</param>
+/// <param name="PayloadId">The backend identifier of the soft-deleted payload row.</param>
+public sealed record PayloadPurgeAck(int PartitionId, long InstanceKey, long PayloadId);

src/Client/Core/TombstonedPayload.cs

@@ -0,0 +1,15 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace Microsoft.DurableTask.Client;
+
+/// <summary>
+/// Serializable representation of a tombstoned payload the backend has soft-deleted and whose blob the
+/// worker should delete. Mirrors the <c>TombstonedPayload</c> protobuf message but is safe to pass through
+/// the orchestration/activity boundary.
+/// </summary>
+/// <param name="PartitionId">The backend partition that owns the payload row.</param>
+/// <param name="InstanceKey">The orchestration instance key the payload belongs to.</param>
+/// <param name="PayloadId">The backend identifier of the soft-deleted payload row.</param>
+/// <param name="Token">The externalized payload token whose backing blob should be deleted.</param>
+public sealed record TombstonedPayload(int PartitionId, long InstanceKey, long PayloadId, string Token);

src/Client/Grpc/GrpcDurableTaskClient.cs

@@ -114,7 +114,7 @@
 
         DateTimeOffset? startAt = options?.StartAt;
         this.logger.SchedulingOrchestration(
             request.InstanceId ?? string.Empty,
             orchestratorName,
             sizeInBytes: request.Input != null ? Encoding.UTF8.GetByteCount(request.Input) : 0,
             startAt.GetValueOrDefault(DateTimeOffset.UtcNow));
@@ -624,6 +624,55 @@
         }
     }
 
+    /// <inheritdoc/>
+    public override async Task<List<TombstonedPayload>> GetTombstonedPayloadsAsync(
+        int limit, CancellationToken cancellation = default)
+    {
+        if (limit <= 0 || limit >= 1000)
+        {
+            throw new ArgumentOutOfRangeException(
+                nameof(limit), limit, "Limit must be greater than 0 and less than 1000.");
+        }
+
+        P.GetTombstonedPayloadsResponse response = await this.sidecarClient.GetTombstonedPayloadsAsync(
+            new P.GetTombstonedPayloadsRequest { Limit = limit },
+            cancellationToken: cancellation);
+
+        List<TombstonedPayload> result = new(response.Payloads.Count);
+        foreach (P.TombstonedPayload payload in response.Payloads)
+        {
+            result.Add(new TombstonedPayload(
+                payload.PartitionId, payload.InstanceKey, payload.PayloadId, payload.Token));
+        }
+
+        return result;
+    }
+
+    /// <inheritdoc/>
+    public override async Task AckPurgedPayloadsAsync(
+        IEnumerable<PayloadPurgeAck> acks, CancellationToken cancellation = default)
+    {
+        Check.NotNull(acks);
+
+        P.AckPurgedPayloadsRequest request = new();
+        foreach (PayloadPurgeAck ack in acks)
+        {
+            request.Acks.Add(new P.PayloadPurgeAck
+            {
+                PartitionId = ack.PartitionId,
+                InstanceKey = ack.InstanceKey,
+                PayloadId = ack.PayloadId,
+            });
+        }
+
+        if (request.Acks.Count == 0)
+        {
+            return;
+        }
+
+        await this.sidecarClient.AckPurgedPayloadsAsync(request, cancellationToken: cancellation);
+    }
+
     static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, ILogger logger, out CallInvoker callInvoker)
     {
         Func<GrpcChannel, CancellationToken, Task<GrpcChannel>>? recreator = options.Internal.ChannelRecreator;

src/Extensions/AzureBlobPayloads/AutoPurge/Activities/AckPurgedPayloadsActivity.cs

@@ -0,0 +1,36 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using Microsoft.DurableTask.Client;
+using Microsoft.Extensions.Logging;
+
+namespace Microsoft.DurableTask.AzureBlobPayloads;
+
+/// <summary>
+/// Activity that acknowledges to the backend the payloads whose blobs the worker has deleted, so the backend
+/// can hard-delete the soft-deleted rows.
+/// </summary>
+/// <param name="client">The Durable Task client used to acknowledge purged payloads to the backend.</param>
+/// <param name="logger">The logger instance.</param>
+[DurableTask]
+public class AckPurgedPayloadsActivity(
+    DurableTaskClient client,
+    ILogger<AckPurgedPayloadsActivity> logger)
+    : TaskActivity<List<PayloadPurgeAck>, object?>
+{
+    readonly DurableTaskClient client = Check.NotNull(client);
+    readonly ILogger<AckPurgedPayloadsActivity> logger = Check.NotNull(logger);
+
+    /// <inheritdoc/>
+    public override async Task<object?> RunAsync(TaskActivityContext context, List<PayloadPurgeAck> input)
+    {
+        if (input is null || input.Count == 0)
+        {
+            return null;
+        }
+
+        await this.client.AckPurgedPayloadsAsync(input, CancellationToken.None);
+        this.logger.BlobPurgeAckedPayloads(input.Count);
+        return null;
+    }
+}

src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs

@@ -0,0 +1,49 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using Microsoft.Extensions.Logging;
+
+namespace Microsoft.DurableTask.AzureBlobPayloads;
+
+/// <summary>
+/// Activity that deletes a single externalized payload blob given its token. Deletion is idempotent, so
+/// re-delivered tokens and concurrent workers are safe. Malformed (poison) tokens are discarded so they get
+/// acknowledged instead of retried forever; transient failures leave the payload tombstoned to retry.
+/// </summary>
+/// <param name="store">The payload store used to delete blobs.</param>
+/// <param name="logger">The logger instance.</param>
+[DurableTask]
+public class DeleteExternalBlobActivity(
+    PayloadStore store,
+    ILogger<DeleteExternalBlobActivity> logger)
+    : TaskActivity<string, BlobDeleteResult>
+{
+    readonly PayloadStore store = Check.NotNull(store);
+    readonly ILogger<DeleteExternalBlobActivity> logger = Check.NotNull(logger);
+
+    /// <inheritdoc/>
+    public override async Task<BlobDeleteResult> RunAsync(TaskActivityContext context, string input)
+    {
+        Check.NotNullOrEmpty(input, nameof(input));
+
+        try
+        {
+            await this.store.DeleteAsync(input, CancellationToken.None);
+            return BlobDeleteResult.Deleted;
+        }
+        catch (ArgumentException ex)
+        {
+            // The token is malformed or points at a different container; it can never succeed. Discard it so
+            // the backend clears the row instead of re-streaming the same poison token every cycle.
+            this.logger.BlobPurgeDeleteDiscarded(ex, input);
+            return BlobDeleteResult.Discarded;
+        }
+        catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException)
+        {
+            // Transient failure: leave the payload tombstoned so a later purge cycle can retry it. A single
+            // bad token must not fail the whole batch.
+            this.logger.BlobPurgeDeleteFailed(ex, input);
+            return BlobDeleteResult.Retry;
+        }
+    }
+}

src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs

@@ -0,0 +1,32 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using Microsoft.DurableTask.Client;
+using Microsoft.Extensions.Logging;
+
+namespace Microsoft.DurableTask.AzureBlobPayloads;
+
+/// <summary>
+/// Activity that fetches a batch of tombstoned payloads from the backend for the auto-purge job to delete.
+/// </summary>
+/// <param name="client">The Durable Task client used to query the backend for tombstoned payloads.</param>
+/// <param name="logger">The logger instance.</param>
+[DurableTask]
+public class GetTombstonedPayloadsActivity(
+    DurableTaskClient client,
+    ILogger<GetTombstonedPayloadsActivity> logger)
+    : TaskActivity<int, List<TombstonedPayload>>
+{
+    readonly DurableTaskClient client = Check.NotNull(client);
+    readonly ILogger<GetTombstonedPayloadsActivity> logger = Check.NotNull(logger);
+
+    /// <inheritdoc/>
+    public override async Task<List<TombstonedPayload>> RunAsync(TaskActivityContext context, int input)
+    {
+        int limit = input > 0 ? input : BlobPurgeConstants.DefaultBatchSize;
+        List<TombstonedPayload> payloads =
+            await this.client.GetTombstonedPayloadsAsync(limit, CancellationToken.None);
+        this.logger.BlobPurgeFetchedTombstones(payloads.Count);
+        return payloads;
+    }
+}

src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs

@@ -0,0 +1,114 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using Microsoft.DurableTask.Client;
+using Microsoft.DurableTask.Entities;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+namespace Microsoft.DurableTask.AzureBlobPayloads;
+
+/// <summary>
+/// Client-side hosted service that ensures the singleton blob payload auto-purge job exists when auto-purge
+/// is enabled via <see cref="LargePayloadStorageOptions.AutoPurge"/>. It never blocks host startup: it runs
+/// on a background task and retries until the backend is reachable. The job is a whole-scheduler singleton,
+/// so racing client processes simply no-op.
+/// </summary>
+sealed class BlobPurgeJobStarter : IHostedService
+{
+    static readonly TimeSpan RetryDelay = TimeSpan.FromSeconds(10);
+
+    readonly DurableTaskClient client;
+    readonly IOptionsMonitor<LargePayloadStorageOptions> options;
+    readonly string builderName;
+    readonly ILogger<BlobPurgeJobStarter> logger;
+    readonly EntityInstanceId entityId = new(nameof(BlobPurgeJob), BlobPurgeConstants.JobId);
+
+    CancellationTokenSource? cts;
+    Task? ensureTask;
+
+    public BlobPurgeJobStarter(
+        DurableTaskClient client,
+        IOptionsMonitor<LargePayloadStorageOptions> options,
+        string builderName,
+        ILogger<BlobPurgeJobStarter> logger)
+    {
+        this.client = Check.NotNull(client);
+        this.options = Check.NotNull(options);
+        this.builderName = Check.NotNull(builderName);
+        this.logger = Check.NotNull(logger);
+    }
+
+    /// <inheritdoc/>
+    public Task StartAsync(CancellationToken cancellationToken)
+    {
+        LargePayloadStorageOptions opts = this.options.Get(this.builderName);
+        if (!opts.AutoPurge)
+        {
+            this.logger.BlobPurgeDisabled();
+            return Task.CompletedTask;
+        }
+
+        int batchSize = opts.PayloadPurgeBatchSize > 0 ? opts.PayloadPurgeBatchSize : BlobPurgeConstants.DefaultBatchSize;
+
+        // Do not block host startup; ensure the job on a background task with basic retry until the backend
+        // is reachable.
+        this.cts = new CancellationTokenSource();
+        this.ensureTask = Task.Run(() => this.EnsureJobAsync(batchSize, this.cts.Token), CancellationToken.None);
+        return Task.CompletedTask;
+    }
+
+    /// <inheritdoc/>
+    public async Task StopAsync(CancellationToken cancellationToken)
+    {
+        this.cts?.Cancel();
+
+        Task? pending = this.ensureTask;
+        if (pending is not null)
+        {
+            // The ensure loop observes cancellation and returns promptly; swallow any faulted/cancelled result.
+            await Task.WhenAny(pending, Task.Delay(Timeout.Infinite, cancellationToken)).ConfigureAwait(false);
+        }
+    }
+
+    async Task EnsureJobAsync(int batchSize, CancellationToken cancellationToken)
+    {
+        while (!cancellationToken.IsCancellationRequested)
+        {
+            try
+            {
+                // The singleton is already guaranteed by the entity's fixed key (Create no-ops when the job is
+                // active) and the orchestrator's fixed instance id, so just schedule the bridge once with a
+                // fixed instance id. Retry only if the backend is unreachable at startup.
+                BlobPurgeJobOperationRequest request = new(
+                    this.entityId, nameof(BlobPurgeJob.Create), batchSize);
+
+                await this.client.ScheduleNewOrchestrationInstanceAsync(
+                    new TaskName(nameof(ExecuteBlobPurgeJobOperationOrchestrator)),
+                    request,
+                    new StartOrchestrationOptions(BlobPurgeConstants.StarterInstanceId),
+                    cancellationToken);
+
+                this.logger.BlobPurgeJobEnsured();
+                return;
+            }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                return;
+            }
+            catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException)
+            {
+                this.logger.BlobPurgeStarterRetry(ex);
+                try
+                {
+                    await Task.Delay(RetryDelay, cancellationToken);
+                }
+                catch (OperationCanceledException)
+                {
+                    return;
+                }
+            }
+        }
+    }
+}