Add large-payload blob auto-purge (opt-in singleton job, worker/SDK side)

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 due large-payload tombstones whose backing blobs a credentialed caller must delete.
+    /// </summary>
+    /// <param name="limit">The maximum number of tombstones to request.</param>
+    /// <param name="cancellation">The cancellation token.</param>
+    /// <returns>The batch of tombstones whose blobs should be deleted.</returns>
+    /// <exception cref="NotSupportedException">Thrown if this implementation does not support the operation.</exception>
+    public virtual Task<List<LargePayloadTombstone>> GetLargePayloadTombstonesAsync(
+        int limit, CancellationToken cancellation = default)
+        => throw new NotSupportedException($"{this.GetType()} does not support retrieving large-payload tombstones.");
+
+    /// <summary>
+    /// Reports the outcome of each attempted large-payload blob deletion so the backend can resolve,
+    /// reschedule, or quarantine the corresponding tombstones. Every attempted row is reported, not just the
+    /// successful ones; the backend owns retry scheduling.
+    /// </summary>
+    /// <param name="results">The per-row outcomes of the attempted deletions.</param>
+    /// <param name="cancellation">The cancellation token.</param>
+    /// <returns>A task that completes when the outcomes have been recorded.</returns>
+    /// <exception cref="NotSupportedException">Thrown if this implementation does not support the operation.</exception>
+    public virtual Task ReportLargePayloadPurgeResultsAsync(
+        IEnumerable<LargePayloadPurgeResult> results, CancellationToken cancellation = default)
+        => throw new NotSupportedException($"{this.GetType()} does not support reporting large-payload purge results.");
+
     // TODO: Create task hub
 
     // TODO: Delete task hub

src/Client/Core/LargePayloadPurgeDisposition.cs

@@ -0,0 +1,33 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace Microsoft.DurableTask.Client;
+
+/// <summary>
+/// The outcome of a single large-payload blob deletion attempt. The split is by whether a failure can
+/// self-heal. Mirrors the <c>LargePayloadPurgeDisposition</c> protobuf enum.
+/// </summary>
+public enum LargePayloadPurgeDisposition
+{
+    /// <summary>
+    /// No disposition was specified.
+    /// </summary>
+    Unspecified = 0,
+
+    /// <summary>
+    /// Terminal success. The blob was deleted, was already absent, or was deliberately left in place because
+    /// it is not owned by the payload store. The backend deletes the tombstone in all three cases.
+    /// </summary>
+    Deleted = 1,
+
+    /// <summary>
+    /// The failure may self-heal, so the row stays pending and the backend sets the next attempt.
+    /// </summary>
+    Retry = 2,
+
+    /// <summary>
+    /// A deterministic failure or protocol violation that retrying can never fix. The backend preserves the
+    /// evidence, alerts, and stops automatic retries.
+    /// </summary>
+    Quarantined = 3,
+}

src/Client/Core/LargePayloadPurgeResult.cs

@@ -0,0 +1,35 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace Microsoft.DurableTask.Client;
+
+/// <summary>
+/// Serializable outcome of exactly one attempted large-payload blob deletion. Mirrors the
+/// <c>LargePayloadPurgeResult</c> protobuf message but is safe to pass through the orchestration/activity
+/// boundary. The backend owns retry scheduling and branches solely on
+/// <see cref="Disposition"/>: it deletes rows reported as
+/// <see cref="LargePayloadPurgeDisposition.Deleted"/>, reschedules
+/// <see cref="LargePayloadPurgeDisposition.Retry"/> on its own backoff, and moves
+/// <see cref="LargePayloadPurgeDisposition.Quarantined"/> rows out of the active fetch. The worker never
+/// computes a retry delay.
+/// </summary>
+/// <remarks>
+/// The disposition is deliberately the only outcome field: anything finer would be write-only on the backend.
+/// Why an attempt failed stays in the worker's own telemetry, which holds the cause at full fidelity rather
+/// than as a lossy classification, and a row is correlated to it by
+/// (<see cref="PartitionId"/>, <see cref="InstanceKey"/>, <see cref="PayloadId"/>).
+/// </remarks>
+/// <param name="PartitionId">The backend partition that owns the tombstoned row.</param>
+/// <param name="InstanceKey">The orchestration instance key the payload belonged to.</param>
+/// <param name="PayloadId">The backend identifier of the tombstoned payload row.</param>
+/// <param name="Revision">
+/// The revision echoed unmodified from the fetched <see cref="LargePayloadTombstone"/>; used by the backend
+/// as a compare-and-swap guard.
+/// </param>
+/// <param name="Disposition">The disposition of the deletion attempt.</param>
+public sealed record LargePayloadPurgeResult(
+    int PartitionId,
+    long InstanceKey,
+    long PayloadId,
+    long Revision,
+    LargePayloadPurgeDisposition Disposition);

src/Client/Core/LargePayloadTombstone.cs

@@ -0,0 +1,23 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace Microsoft.DurableTask.Client;
+
+/// <summary>
+/// Serializable representation of a tombstoned large-payload row whose external blob a credentialed caller
+/// must delete. Mirrors the <c>LargePayloadTombstone</c> protobuf message but is safe to pass through the
+/// orchestration/activity boundary.
+/// </summary>
+/// <param name="PartitionId">The backend partition that owns the tombstoned row.</param>
+/// <param name="InstanceKey">The orchestration instance key the payload belonged to.</param>
+/// <param name="PayloadId">The backend identifier of the tombstoned payload row.</param>
+/// <param name="Token">
+/// The self-describing <c>blob:v2:{fullBlobUrl}</c> payload token whose backing blob should be deleted.
+/// </param>
+/// <param name="Revision">
+/// An optimistic-concurrency guard echoed back unmodified in the corresponding
+/// <see cref="LargePayloadPurgeResult"/> so the backend can reject duplicate or stale reports without taking
+/// a per-row lease.
+/// </param>
+public sealed record LargePayloadTombstone(
+    int PartitionId, long InstanceKey, long PayloadId, string Token, long Revision);

src/Client/Grpc/GrpcDurableTaskClient.cs

@@ -624,6 +624,82 @@ public override async Task<IList<HistoryEvent>> GetOrchestrationHistoryAsync(
         }
     }
 
+    /// <inheritdoc/>
+    public override async Task<List<LargePayloadTombstone>> GetLargePayloadTombstonesAsync(
+        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.GetLargePayloadTombstonesResponse response;
+        try
+        {
+            response = await this.sidecarClient.GetLargePayloadTombstonesAsync(
+                new P.GetLargePayloadTombstonesRequest { Limit = limit },
+                cancellationToken: cancellation);
+        }
+        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
+        {
+            throw new OperationCanceledException(
+                $"The {nameof(this.GetLargePayloadTombstonesAsync)} operation was canceled.", e, cancellation);
+        }
+
+        List<LargePayloadTombstone> result = new(response.Tombstones.Count);
+        foreach (P.LargePayloadTombstone tombstone in response.Tombstones)
+        {
+            result.Add(new LargePayloadTombstone(
+                tombstone.PartitionId,
+                tombstone.InstanceKey,
+                tombstone.PayloadId,
+                tombstone.Token,
+                tombstone.Revision));
+        }
+
+        return result;
+    }
+
+    /// <inheritdoc/>
+    public override async Task ReportLargePayloadPurgeResultsAsync(
+        IEnumerable<LargePayloadPurgeResult> results, CancellationToken cancellation = default)
+    {
+        Check.NotNull(results);
+
+        P.ReportLargePayloadPurgeResultsRequest request = new();
+        foreach (LargePayloadPurgeResult result in results)
+        {
+            request.Results.Add(new P.LargePayloadPurgeResult
+            {
+                PartitionId = result.PartitionId,
+                InstanceKey = result.InstanceKey,
+                PayloadId = result.PayloadId,
+                Revision = result.Revision,
+
+                // The managed disposition enum declares the same numeric values as its protobuf counterpart,
+                // so it maps across by value. This is the only enum on the message and it only travels
+                // outbound, so the SDK can never receive a value it does not know.
+                Disposition = (P.LargePayloadPurgeDisposition)result.Disposition,
+            });
+        }
+
+        if (request.Results.Count == 0)
+        {
+            return;
+        }
+
+        try
+        {
+            await this.sidecarClient.ReportLargePayloadPurgeResultsAsync(request, cancellationToken: cancellation);
+        }
+        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
+        {
+            throw new OperationCanceledException(
+                $"The {nameof(this.ReportLargePayloadPurgeResultsAsync)} operation was canceled.", e, cancellation);
+        }
+    }
+
     static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, ILogger logger, out CallInvoker callInvoker)
     {
         Func<GrpcChannel, CancellationToken, Task<GrpcChannel>>? recreator = options.Internal.ChannelRecreator;