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

@@ -624,6 +624,72 @@ public override async Task<IList<HistoryEvent>> GetOrchestrationHistoryAsync(
         }
     }
 
+    /// <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 or equal to 1000.");
+        }
+
+        P.GetTombstonedPayloadsResponse response;
+        try
+        {
+            response = await this.sidecarClient.GetTombstonedPayloadsAsync(
+                new P.GetTombstonedPayloadsRequest { Limit = limit },
+                cancellationToken: cancellation);
+        }
+        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
+        {
+            throw new OperationCanceledException(
+                $"The {nameof(this.GetTombstonedPayloadsAsync)} operation was canceled.", e, 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;
+        }
+
+        try
+        {
+            await this.sidecarClient.AckPurgedPayloadsAsync(request, cancellationToken: cancellation);
+        }
+        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
+        {
+            throw new OperationCanceledException(
+                $"The {nameof(this.AckPurgedPayloadsAsync)} operation was canceled.", e, 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,81 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using Azure;
+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.
+/// </summary>
+/// <remarks>
+/// Outcome classification, verified against the Azure.Storage.Blobs / Azure.Core exception model (not
+/// assumed):
+/// <list type="bullet">
+/// <item>
+/// The Azure SDK already retries transient failures internally (connection errors plus HTTP
+/// 408/429/500/502/503/504, with exponential backoff), so any exception that escapes
+/// <see cref="PayloadStore.DeleteAsync"/> means those built-in retries were already exhausted.
+/// </item>
+/// <item>
+/// Permanent failures are discarded (acked so the backend clears the row): an <see cref="ArgumentException"/>
+/// from the store's own token decode / container-mismatch check (thrown client-side before any network call),
+/// and a <see cref="RequestFailedException"/> with <see cref="RequestFailedException.Status"/> 400 (for
+/// example InvalidUri / InvalidResourceName when the decoded blob name violates Azure naming rules). Retrying
+/// either can never succeed.
+/// </item>
+/// <item>
+/// Everything else is treated as transient and leaves the payload tombstoned to retry on a later cycle:
+/// throttling / 5xx that outlived the SDK's retries, 403 authorization failures (which need an operator
+/// credential fix rather than dropping data), and timeouts / cancellation. A blob is never dropped on an
+/// uncertain error, and a single bad token never fails the whole batch.
+/// </item>
+/// </list>
+/// </remarks>
+/// <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 (RequestFailedException ex) when (ex.Status == 400)
+        {
+            // Service rejected the request as permanently invalid (e.g. InvalidUri / InvalidResourceName - the
+            // decoded blob name violates Azure naming rules). Retrying can never succeed, so discard it like a
+            // poison token: ack so the backend clears the row instead of re-streaming it forever.
+            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;
+        List<TombstonedPayload> payloads =
+            await this.client.GetTombstonedPayloadsAsync(limit, CancellationToken.None);
+        this.logger.BlobPurgeFetchedTombstones(payloads.Count);
+        return payloads;
+    }
+}