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.

Suppressed comments (1)

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

• BlobPurgeJobState.LastError is documented as “the last error message”, but it’s never set anywhere in the auto-purge implementation (only cleared on Create). This is misleading for anyone reading the entity state; either wire it up (e.g., set it on cycle failures) or update the docs to reflect that it’s currently unused/reserved.

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

[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/Orchestrations/BlobPurgeJobOrchestrator.cs:47

• PurgeBatchSize can be persisted as 0 (see BlobPurgeJobTests) and is then passed through to GetLargePayloadTombstonesActivity, which calls DurableTaskClient.GetLargePayloadTombstonesAsync(limit). The gRPC client throws ArgumentOutOfRangeException when limit <= 0, causing the orchestrator to repeatedly fail/back off and never make progress. Add a defensive clamp to a valid range before using the value.

        string jobId = input.JobEntityId.Key;

        int batchSize = input.PurgeBatchSize;
        int processedCycles = input.ProcessedCycles;

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

• StopAsync waits for the background ensure task but never observes its completion/exception. If EnsureJobAsync ever faults unexpectedly, the exception may surface later as an UnobservedTaskException. After WhenAny returns, await the task (inside a try/catch) when it completed to observe and swallow the result as intended.

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

[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/Entity/BlobPurgeJob.cs:36

• BlobPurgeJob.Create currently accepts any purgeBatchSize (including 0/negative), stores it, and signals the purge orchestrator. If a client ever creates the job with an invalid batch size, the orchestrator will later call GetLargePayloadTombstonesActivity with that value, and GrpcDurableTaskClient.GetLargePayloadTombstonesAsync throws for limit <= 0, causing the job to spin in the error/backoff loop. Validate the batch size at the entity boundary to keep the job from entering an unrecoverable misconfigured state.

    public void Create(TaskEntityContext context, int purgeBatchSize)
    {
        if (this.State.Status == BlobPurgeJobStatus.Active)
        {
            logger.BlobPurgeJobAlreadyRunning(context.Id.Key);
            return;
        }

        this.State.Status = BlobPurgeJobStatus.Active;
        this.State.PurgeBatchSize = purgeBatchSize;
        this.State.CreatedAt ??= DateTimeOffset.UtcNow;
        this.State.LastModifiedAt = DateTimeOffset.UtcNow;
        this.State.LastError = null;

test/Extensions/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs:78

• This test currently asserts that BlobPurgeJob.Create accepts a batch size of 0. If Create is callable via the client-to-entity bridge, persisting an invalid batch size can wedge the job (the gRPC client rejects limit <= 0). Update the test to assert the expected validation behavior instead.

    [Fact]
    public async Task Create_StoresBatchSizeVerbatim_WithoutCoercion()
    {
        // Arrange - the batch size is validated once at specification (LargePayloadStorageOptions), so the
        // entity trusts its input and performs no coercion of its own. A zero here is stored as-is, proving
        // the previous non-positive-to-default fallback was removed.
        TestEntityOperation operation = new(
            nameof(BlobPurgeJob.Create),
            new TestEntityState(null),
            0);

        // Act
        await this.job.RunAsync(operation);

        // Assert
        BlobPurgeJobState state = Assert.IsType<BlobPurgeJobState>(
            operation.State.GetState(typeof(BlobPurgeJobState)));
        state.PurgeBatchSize.Should().Be(0);
    }

[Copilot]

Pull request overview

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

Suppressed comments (2)

src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs:28

• BlobPurgeJob.Create stores purgeBatchSize verbatim with no range validation. If an invalid value (<=0 or > MaxBatchSize) is ever passed (e.g., direct entity call), the scheduled BlobPurgeJobOrchestrator will repeatedly fail when fetching tombstones (client validates 1..1000) and the job will back off forever. Validate the range at the entity boundary and fail fast with ArgumentOutOfRangeException.

    public void Create(TaskEntityContext context, int purgeBatchSize)
    {
        if (this.State.Status == BlobPurgeJobStatus.Active)
        {
            logger.BlobPurgeJobAlreadyRunning(context.Id.Key);

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

• DeleteExternalBlobActivity claims it returns failures as a disposition rather than throwing, but RunAsync currently throws on empty input via Check.NotNullOrEmpty. Since the token comes from the backend, treating an empty token as a deterministic protocol failure and returning a Quarantined disposition keeps a single bad row from failing the whole batch and avoids an activity retry loop that never produces a per-row result.

    public override async Task<BlobPurgeOutcome> RunAsync(TaskActivityContext context, string input)
    {
        Check.NotNullOrEmpty(input, nameof(input));

        return await this.DeleteAsync(input);

[wangbill (YunchuWang)]

move to constant

[wangbill (YunchuWang)]

same here, we need to handle other exception cases as well

[wangbill (YunchuWang)]

can we parse the token into payload token format, if version number is < v2, we dont support auto purge and so versions > v2 we support. if token does not follow the format defined, should be considered deleted and delete it from dts backend as well

[wangbill (YunchuWang)]

isV2 as a flag is hacky, should be something like autoPurgable?

[Copilot]

Pull request overview

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

Suppressed comments (3)

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

• The auto-purge startup gate hard-codes BlobPayloadStore as the only delete-capable store (if (this.store is not BlobPayloadStore)), which contradicts the new PayloadStore.DeleteAsync contract (virtual, intended for other implementations to override). This prevents a custom PayloadStore that overrides DeleteAsync from ever enabling auto-purge, even if it can delete the v2 blob tokens correctly. Consider gating on whether DeleteAsync is overridden (or otherwise supported) rather than on the concrete type.

        if (this.store is not BlobPayloadStore)
        {
            this.logger.BlobPurgeStoreCannotDelete(this.store.GetType().FullName);
            return Task.CompletedTask;
        }

src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs:62

• Run passes this.State.PurgeBatchSize straight into BlobPurgeJobRunRequest without range validation. If the entity is created with an out-of-range batch size (e.g. 0), GetLargePayloadTombstonesAsync will throw ArgumentOutOfRangeException (limit must be 1..1000) and the orchestrator will back off/retry forever. Clamping or defaulting the batch size at scheduling time avoids a permanently bricked job state.

        context.ScheduleNewOrchestration(
            new TaskName(nameof(BlobPurgeJobOrchestrator)),
            new BlobPurgeJobRunRequest(context.Id, this.State.PurgeBatchSize),
            startOrchestrationOptions);

src/Client/Grpc/GrpcDurableTaskClient.cs:685

• ReportLargePayloadPurgeResultsAsync blindly casts result.Disposition to the protobuf enum. If a caller passes the default LargePayloadPurgeDisposition.Unspecified (or any undefined numeric value), the request can be rejected server-side and potentially break purge reporting. Validating the disposition before adding it to the request makes misuse fail fast with a clear client-side error.

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

[Copilot]

Pull request overview

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

Suppressed comments (4)

src/Client/Grpc/GrpcDurableTaskClient.cs:685

• ReportLargePayloadPurgeResultsAsync currently allows LargePayloadPurgeDisposition.Unspecified to be sent over the wire via the numeric cast. The proto explicitly makes 0 a meaningless sentinel and says the backend must reject it; validating here would fail fast with a clearer exception and avoid a round-trip RPC failure.

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

src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs:63

• TryAddSingleton registers a single PayloadStore for the entire IServiceCollection but constructs it using named options for the current builder (monitor.Get(builder.Name)). If multiple named clients/workers call UseExternalizedPayloads with different builder names, later builders will reuse a PayloadStore constructed for a different name, while their PostConfigure path still reads options for their own name. That can lead to interceptor/store configuration mismatches (e.g., container/account settings) depending on registration order.

        builder.Services.TryAddSingleton<PayloadStore>(sp =>
        {
            LargePayloadStorageOptions opts = sp.GetRequiredService<IOptionsMonitor<LargePayloadStorageOptions>>().Get(builder.Name);
            return new BlobPayloadStore(opts);
        });

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

• Same concern as the client builder extension: TryAddSingleton creates at most one PayloadStore but builds it from monitor.Get(builder.Name). With multiple named workers (or mixed named worker+client) using different names, the singleton store may be constructed from the wrong named options instance, producing mismatched configuration relative to the named GrpcDurableTaskWorkerOptions post-configure path.

        builder.Services.TryAddSingleton<PayloadStore>(sp =>
        {
            LargePayloadStorageOptions opts = sp.GetRequiredService<IOptionsMonitor<LargePayloadStorageOptions>>().Get(builder.Name);
            return new BlobPayloadStore(opts);
        });

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

• StopAsync waits for the background ensure task to complete, but it never observes task faults. If EnsureJobAsync ever throws unexpectedly (e.g., new exception type added later), this can surface as an UnobservedTaskException. Consider observing and swallowing the exception when the task completes during StopAsync.

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

[Copilot]

Pull request overview

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