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

[wangbill (YunchuWang)]

Summary

Large orchestration payloads are externalized to Azure Blob Storage by the AzureBlobPayloads extension, with a token persisted in SQL instead of the payload bytes. When the orchestration is purged, DTS removes the SQL state but cannot delete the backing blob — it has no customer storage credentials. Only the worker does.

This PR implements the worker/SDK side. The backend records a durable tombstone for each externalized payload whose orchestration state is gone; the worker fetches due tombstones, deletes the blobs, and reports the outcome of every row so the backend can resolve, reschedule, or quarantine it.

Companion changes: contract in microsoft/durabletask-protobuf#76, backend in AAPT-DTMB PR 16368738.

gRPC contract

Two unary RPCs on TaskHubSidecarService (worker is the client). The vendored src/Grpc/orchestrator_service.proto is byte-identical to protobuf#76 — verified mechanically by exact-substring comparison, not by eye.

rpc GetLargePayloadTombstones(GetLargePayloadTombstonesRequest) returns (GetLargePayloadTombstonesResponse);
rpc ReportLargePayloadPurgeResults(ReportLargePayloadPurgeResultsRequest) returns (ReportLargePayloadPurgeResultsResponse);

• LargePayloadTombstone { partitionId, instanceKey, payloadId, token, revision }
• LargePayloadPurgeResult { identity, revision, disposition }
• Opt-in is google.protobuf.BoolValue large_payload_auto_purge_enabled = 12 on the existing GetWorkItemsRequest — no new handshake, and null means "no opinion".

revision is echoed back unmodified as a compare-and-swap guard, so duplicate or stale reports are no-ops without a per-row lease.

Dispositions

Three dispositions, split on whether a failure can self-heal. There is no Discarded: a success is reported explicitly as Deleted, and every failure is carried by Retry or Quarantined.

Disposition	Meaning	Examples
Deleted	Tombstone resolved	blob deleted; already absent; blob not store-owned
Retry	Transient; backend reschedules	storage unreachable, 401/403, 408/429/5xx, unknown token prefix
Quarantined	Deterministic; retrying can never succeed	malformed v2 token, HTTP 400, legacy v1 token

The worker never computes a retry delay. It reports the failure and the backend owns scheduling and backoff. The orchestrator reports the whole batch unconditionally — including retryable rows — because the backend needs to hear about a failure in order to defer the row. If every row comes back Retry (a storage outage), the cycle applies ErrorBackoff so an outage cannot become a tight refetch loop.

disposition is the entire outcome. An earlier revision also carried reason and storageErrorCode; both were removed after verifying they were write-only end to end — the backend persists them and nothing reads either one. Failure detail is logged by the worker instead, at the classification site, which is strictly richer than the enum was: 401/403, 5xx, and an unreachable account are distinct in telemetry where the enum collapsed all three into one value.

Blob ownership marker

A recognized token proves only that the text looks like one this store emits — the column is customer-writable. So ownership is recorded on the object itself: UploadAsync writes fixed blob metadata managed_by=dts, and the worker re-reads the target's metadata immediately before deleting.

• Marker present → delete, using If-Match on the ETag from that same read, so a mid-flight overwrite fails the delete rather than destroying newer content.
• Marker absent → blob left untouched, and the row is still reported Deleted, so the tombstone is resolved rather than retried forever. This is an expected outcome, not a defect, and must not be quarantined. The worker logs it distinctly so a customer whose payloads are all self-authored is still visible in telemetry.

The metadata name uses an underscore because Azure requires blob metadata names to be valid C# identifiers; managed-by would be rejected at upload.

Only blob:v2: tokens are auto-purged. A v1 token reaching this path is an invariant violation (v1 is excluded at insertion) and is quarantined rather than deleted.

Testing

Verified on a clean (--no-incremental) build:

• dotnet build Microsoft.DurableTask.sln — 0 errors
• test/Extensions/AzureBlobPayloads.Tests — 54 passed
• test/Client/Grpc.Tests — 56 passed
• Blame-attributed warnings — 127 unique in-repo, exactly 1 introduced by this branch

LargePayloadPurgeEnumParityTests pins proto↔managed parity for LargePayloadPurgeDisposition by value and name in both directions, and asserts that no inbound type exposes an enum. Disposition is the one enum crossing the wire, and its numeric cast in GrpcDurableTaskClient is what decides whether a row is deleted or quarantined; that cast is safe only because enums travel outbound-only, and the test fails the build if a future change breaks that invariant.

Notes / intentional deviations

• Ships one intentional warning: CA1873 at BlobPurgeJobOrchestrator.cs:66 (unguarded logging). Kept deliberately for consistency with 77 existing instances across the solution. Blame-based attribution against main confirms this is the only warning this branch introduces.
• PurgedCount counts every Deleted row, including blobs skipped for lacking the ownership marker. Excluding them would pin the counter at 0 for a customer whose payloads are all self-authored, making a healthy draining job read as wedged; the worker log supplies the precision instead.
• PayloadStore.DeleteAsync is virtual with a default that throws NotSupportedException, so existing external subclasses are unaffected.
• BlobPurgeJob.Create is a no-op when already Active so racing client processes don't disturb a running job.
• BlobPurgeJobStarter implements IDisposable rather than disposing its CTS in StopAsync: that method returns on the host shutdown token while the ensure task may still be live, so disposing there would fault it with an unobserved ObjectDisposedException.

Depends on protobuf#76 for the authoritative contract and on the DTS backend serving these RPCs.

[Copilot]

Pull request overview

Copilot reviewed 36 out of 36 changed files in this pull request and generated no new comments.

[Copilot]

Pull request overview

Copilot reviewed 36 out of 36 changed files in this pull request and generated no new comments.

Suppressed comments (2)

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

• StopAsync waits for the background task to complete, but it never observes the task result. If EnsureJobAsync/SignalJobStopAsync faults, the exception can remain unobserved and surface later via UnobservedTaskException (or be lost), making failures harder to diagnose.

Consider awaiting the task when it completes first (and swallowing exceptions intentionally) so any fault is observed deterministically.

            // The background loop observes cancellation and returns promptly; swallow any faulted/cancelled
            // result.
            await Task.WhenAny(pending, Task.Delay(Timeout.Infinite, cancellationToken)).ConfigureAwait(false);
        }

src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs:87

• cycleBatchSize can become 0 when both the stored batch size and the orchestrator input are non-positive (e.g., an older/cleared entity state, or an external call that created the job with 0). In that case GetLargePayloadTombstonesActivity will end up calling DurableTaskClient.GetLargePayloadTombstonesAsync(0), which throws ArgumentOutOfRangeException and will put the orchestrator into an infinite "cycle failed" + backoff loop.

To keep the job self-healing, consider clamping the effective batch size to a safe default when both sources are unset.

                // Fall back to the input when the stored value is not positive. An entity written by an older
                // build carries no batch size at all, and asking the backend for zero rows every cycle would be
                // a silent, permanent stall.
                int cycleBatchSize = state.PurgeBatchSize > 0 ? state.PurgeBatchSize : batchSize;

[Copilot]

Pull request overview

Copilot reviewed 36 out of 36 changed files in this pull request and generated no new comments.

[Copilot]

Pull request overview

Copilot reviewed 36 out of 36 changed files in this pull request and generated no new comments.

Suppressed comments (1)

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

• StopAsync waits with Task.WhenAny but never observes/awaits the background task when it completes, so if EnsureJobAsync/SignalJobStopAsync faults the exception can remain unobserved (risking UnobservedTaskException and making failures harder to diagnose). Consider awaiting the task (in a try/catch) when it wins the race, while still returning early when the host shutdown token fires.

        {
            // The background loop observes cancellation and returns promptly; swallow any faulted/cancelled
            // result.
            await Task.WhenAny(pending, Task.Delay(Timeout.Infinite, cancellationToken)).ConfigureAwait(false);
        }

[Copilot]

Pull request overview

Copilot reviewed 36 out of 36 changed files in this pull request and generated no new comments.

Suppressed comments (2)

src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobState.cs:41

• BlobPurgeJobState.LastError is documented as carrying the last error message, but nothing in the auto-purge implementation ever sets it (it’s only cleared in BlobPurgeJob.Create). Either wire this up (e.g., an entity operation the orchestrator calls in its catch-all) or adjust the documentation so consumers don’t assume it will be populated.

    /// <summary>
    /// Gets or sets the last error message, if any.
    /// </summary>
    public string? LastError { get; set; }

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

• StopAsync waits for the background task to finish but never observes its result. If the background task faults, this can leave an unobserved exception (and it also makes it harder to diagnose failures). Consider awaiting the task when it completes, swallowing exceptions only after they’ve been observed.

        {
            // The background loop observes cancellation and returns promptly; swallow any faulted/cancelled
            // result.
            await Task.WhenAny(pending, Task.Delay(Timeout.Infinite, cancellationToken)).ConfigureAwait(false);
        }

[Copilot]

Pull request overview

Copilot reviewed 36 out of 36 changed files in this pull request and generated no new comments.

Suppressed comments (2)

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

• EnsureJobAsync captures batchSize once at StartAsync and then reuses it forever, so runtime configuration updates (IOptionsMonitor) won't ever reach the job entity. Since reconciliation runs for the lifetime of the process, re-read the current PayloadPurgeBatchSize each pass before scheduling the bridge orchestration.

            try
            {
                BlobPurgeJobOperationRequest request = new(
                    this.entityId, nameof(BlobPurgeJob.Create), batchSize);

src/Client/Grpc/GrpcDurableTaskClient.cs:675

• ReportLargePayloadPurgeResultsAsync will happily send a result whose Disposition is Unspecified (0). Since the proto explicitly requires 0 to be rejected by the backend, validate client input and throw before sending to avoid backend-side failures (or accidental misreporting if a future backend changes behavior).

        foreach (LargePayloadPurgeResult result in results)
        {
            request.Results.Add(new P.LargePayloadPurgeResult
            {
                PartitionId = result.PartitionId,

[wangbill (YunchuWang)]

v1 token should not even be in tomberstone, backend only put v2 tokens to tomberstone

[wangbill (YunchuWang)]

rename to notdtsowned

[wangbill (YunchuWang)]

move this check outside the activity to reduce activity call if not needed

[Copilot]

Pull request overview

Copilot reviewed 36 out of 36 changed files in this pull request and generated no new comments.

Suppressed comments (1)

src/Client/Grpc/GrpcDurableTaskClient.cs:701

• ReportLargePayloadPurgeResultsAsync will happily serialize and send LargePayloadPurgeDisposition.Unspecified (0) (or any out-of-range enum value) via a raw numeric cast. The proto contract explicitly requires 0 to be rejected by the backend; validating here gives callers an immediate, clear exception and avoids a round-trip that can only fail server-side.

    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,
            });
        }