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 29 out of 29 changed files in this pull request and generated no new comments.

Suppressed comments (2)

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

• purgeBatchSize is stored verbatim into entity state without validation. If a caller invokes BlobPurgeJob.Create with 0/negative or >1000, the orchestrator will later fail when calling GetTombstonedPayloads (the gRPC client enforces 1..1000), causing the job to back off and loop forever without making progress. Consider validating the range here (or coercing to a safe default) even if the starter path uses validated options, since the entity operation is callable independently.

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

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

• The auto-purge capability gate is currently this.store is not BlobPayloadStore, which assumes only BlobPayloadStore can delete payloads. Since PayloadStore.DeleteAsync was intentionally added as a virtual extensibility point, a custom PayloadStore could support deletion without being BlobPayloadStore. Consider gating on whether DeleteAsync is overridden (capability) rather than the concrete store type (implementation), so delete-capable custom stores can still opt in.

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

[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:55

• processedCycles is incremented before the continue-as-new check, but the condition uses >; this causes the orchestrator to ContinueAsNew after 6 cycles when ContinueAsNewFrequency is 5. This contradicts the constant’s intent and increases history size more than necessary.

            processedCycles++;
            if (processedCycles > ContinueAsNewFrequency)
            {
                context.ContinueAsNew(new BlobPurgeJobRunRequest(input.JobEntityId, batchSize, ProcessedCycles: 0));
                return null!;

src/Grpc/orchestrator_service.proto:799

• PR description and the embedded gRPC contract snippet describe GetTombstonedPayloads / AckPurgedPayloads, but this proto defines GetLargePayloadTombstones / ReportLargePayloadPurgeResults. Before merging (and before syncing from the upstream protobuf repo), please reconcile the RPC/message naming to avoid a breaking drift between the SDK and the authoritative proto.

    // Returns a bounded, deterministically ordered batch of due large-payload tombstones whose
    // external blobs the worker must delete. Scoped to the caller's authenticated task hub.
    // Only rows that are pending and whose next attempt time has arrived are returned; a row stays
    // pending until its outcome is reported, so this is safe under retries and duplicate callers.
    rpc GetLargePayloadTombstones(GetLargePayloadTombstonesRequest) returns (GetLargePayloadTombstonesResponse);

    // Reports the outcome of each attempted blob deletion. The backend owns retry scheduling: it
    // deletes rows reported as DELETED, reschedules RETRY with a reason-appropriate next attempt,
    // and moves QUARANTINED rows out of the active fetch while preserving their evidence.
    rpc ReportLargePayloadPurgeResults(ReportLargePayloadPurgeResultsRequest) returns (ReportLargePayloadPurgeResultsResponse);

[Copilot]

Pull request overview

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

Suppressed comments (3)

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

• BlobPurgeJob.Create stores purgeBatchSize verbatim without validation. Because the entity can be invoked via the public bridge orchestrator, an invalid value (e.g., 0) can be persisted and later cause GetLargePayloadTombstonesAsync(limit) to throw (limit must be 1..1000), resulting in a perpetual error/backoff loop.

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

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

• ContinueAsNewFrequency is documented/used as "every 5 cycles", but the condition processedCycles > ContinueAsNewFrequency triggers on the 6th cycle (processedCycles is incremented before the check). This keeps more history than intended.

            processedCycles++;
            if (processedCycles > ContinueAsNewFrequency)
            {
                context.ContinueAsNew(new BlobPurgeJobRunRequest(input.JobEntityId, batchSize, ProcessedCycles: 0));
                return null!;
            }

src/Grpc/orchestrator_service.proto:799

• The PR description's "gRPC contract" section describes unary RPCs named GetTombstonedPayloads/AckPurgedPayloads with different message shapes, but the vendored proto here defines GetLargePayloadTombstones and ReportLargePayloadPurgeResults (with additional disposition/reason enums). This makes it hard to reconcile the intended wire contract and increases the risk of incompatibility with the upstream protobuf PR.

    // Returns a bounded, deterministically ordered batch of due large-payload tombstones whose
    // external blobs the worker must delete. Scoped to the caller's authenticated task hub.
    // Only rows that are pending and whose next attempt time has arrived are returned; a row stays
    // pending until its outcome is reported, so this is safe under retries and duplicate callers.
    rpc GetLargePayloadTombstones(GetLargePayloadTombstonesRequest) returns (GetLargePayloadTombstonesResponse);

    // Reports the outcome of each attempted blob deletion. The backend owns retry scheduling: it
    // deletes rows reported as DELETED, reschedules RETRY with a reason-appropriate next attempt,
    // and moves QUARANTINED rows out of the active fetch while preserving their evidence.
    rpc ReportLargePayloadPurgeResults(ReportLargePayloadPurgeResultsRequest) returns (ReportLargePayloadPurgeResultsResponse);

[Copilot]

Pull request overview

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

Suppressed comments (4)

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

• BlobPurgeJob.Create stores purgeBatchSize verbatim into entity state. If a client ever calls Create with 0/negative (or an excessively large value), the orchestrator will pass it to GetLargePayloadTombstonesActivity/GrpcDurableTaskClient.GetLargePayloadTombstonesAsync, which throws for limit <= 0 || limit > 1000 (GrpcDurableTaskClient.cs:631-635). That would put the job into an endless backoff loop. Consider validating/clamping purgeBatchSize here to keep the entity from entering an unrecoverable configuration state.

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

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

• BlobPurgeJobStarter gates auto-purge on store is BlobPayloadStore, but PayloadStore.DeleteAsync was introduced as a virtual extensibility point. This type-check prevents any custom PayloadStore that overrides DeleteAsync (and can safely delete its own payload objects) from enabling auto-purge, even though the API suggests it should be supported. Consider gating on whether DeleteAsync is actually 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/Client/BlobPurgeJobStarter.cs:109

• StopAsync waits for either the ensure task or the host shutdown token, but it never observes ensureTask's completion result. If the background task faults, the exception can go unobserved (raising UnobservedTaskException later) and makes debugging harder. Consider awaiting the task when it completes (swallowing exceptions as intended) to ensure faults are observed.

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

src/Grpc/orchestrator_service.proto:795

• The PR description documents new sidecar RPC names GetTombstonedPayloads / AckPurgedPayloads, but the vendored proto defines GetLargePayloadTombstones / ReportLargePayloadPurgeResults (and the client APIs follow those names). Please update the PR description (or rename the RPCs) so the documented contract matches the code.

    rpc GetLargePayloadTombstones(GetLargePayloadTombstonesRequest) returns (GetLargePayloadTombstonesResponse);

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