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

[YunchuWang]

Summary

Large orchestration payloads are externalized to Azure Blob Storage by the AzureBlobPayloads extension as blob:v1:<container>:<blobName> tokens. The DTS backend stores those tokens but cannot delete the backing blobs — it has no storage credentials; only this SDK can. This PR implements the worker/SDK side of large-payload blob auto-purge.

Design (opt-in singleton durable entity + orchestration job)

Instead of an always-on background stream, this mirrors the existing src/ExportHistory feature: an opt-in, whole-scheduler singleton durable entity + orchestration job that drains soft-deleted payload rows the backend exposes and deletes their blobs, then acks so the backend can hard-delete the rows.

• PayloadStore.DeleteAsync — added as a virtual method (default throws NotSupportedException, so it is non-breaking for existing external subclasses). BlobPayloadStore overrides it to decode the token and call DeleteIfExistsAsync (idempotent — deleting a missing blob is a no-op).
• BlobPurgeJob (TaskEntity singleton) — Create is a no-op when already Active so racing client processes don't disturb the running job (intentionally softer than ExportJob.Create, which throws). Run starts a fixed-instance-id orchestrator.
• BlobPurgeJobOrchestrator (perpetual) — each cycle: fetch a batch of tombstones, delete the blobs with capped parallelism (32), ack only the successful deletions (failed tokens stay tombstoned to retry), idle on a 1-minute timer when there's nothing to purge, and ContinueAsNew every 5 cycles to keep history small. Activities use a small retry policy.
• ExecuteBlobPurgeJobOperationOrchestrator — client → entity bridge (mirrors export).
• Activities (constructor DI): GetTombstonedPayloadsActivity, DeleteExternalBlobActivity (returns false + logs on failure so one bad token can't fail the batch), AckPurgedPayloadsActivity.
• Purge RPCs on the core client — the worker fetch/ack RPCs are now first-class methods on DurableTaskClient (GetTombstonedPayloadsAsync/AckPurgedPayloadsAsync), overridden in GrpcDurableTaskClient — mirroring how ExportHistory added ListInstanceIdsAsync/GetOrchestrationHistoryAsync. The purge activities inject DurableTaskClient directly; no dedicated gRPC client is needed. AzureManaged reuses GrpcDurableTaskClient, so it inherits these methods with no extra client.
• Options — LargePayloadStorageOptions.AutoPurge (opt-in, default false) and PayloadPurgeBatchSize (default 500).
• Enablement — client-side BlobPurgeJobStarter (IHostedService) ensures the singleton job when AutoPurge is enabled, on a background task that does not block host startup and retries until the backend is reachable. The worker always registers the entity/orchestrators/activities (not gated on AutoPurge) so a client-enabled job always has something to execute.

gRPC contract

Two new unary RPCs added to TaskHubSidecarService in src/Grpc/orchestrator_service.proto (worker is the client; wire paths /TaskHubSidecarService/GetTombstonedPayloads and /AckPurgedPayloads):

rpc GetTombstonedPayloads(GetTombstonedPayloadsRequest) returns (GetTombstonedPayloadsResponse);
rpc AckPurgedPayloads(AckPurgedPayloadsRequest) returns (AckPurgedPayloadsResponse);

message TombstonedPayload { int32 partitionId = 1; int64 instanceKey = 2; int64 payloadId = 3; string token = 4; }
message PayloadPurgeAck   { int32 partitionId = 1; int64 instanceKey = 2; int64 payloadId = 3; }
message GetTombstonedPayloadsRequest  { int32 limit = 1; }
message GetTombstonedPayloadsResponse { repeated TombstonedPayload payloads = 1; }
message AckPurgedPayloadsRequest      { repeated PayloadPurgeAck acks = 1; }
message AckPurgedPayloadsResponse     { }

C# stubs are generated at build time by Grpc.Tools (not committed). The authoritative proto change is a follow-up in microsoft/durabletask-protobuf#76; once it merges, src/Grpc/orchestrator_service.proto should be re-synced from upstream (content identical to what's here). The vendored change lets this PR build/test standalone.

Testing

• dotnet build Microsoft.DurableTask.sln — succeeds, 0 errors.
• dotnet test test/AzureBlobPayloads.Tests — 11 passed (BlobPayloadStore delete/idempotency + BlobPurgeJob.Create no-op-when-Active + options defaults).
• dotnet test test/ExportHistory.Tests — 147 passed (shared patterns unaffected).
• dotnet test test/Client/Core.Tests + test/Client/Grpc.Tests — 43 + 47 passed (core DurableTaskClient / GrpcDurableTaskClient edits).

Notes / deviations

• BlobPurgeJobStatus.Pending is the 0/default value (mirroring ExportJobStatus.Pending=0) so a freshly initialized entity never appears Active.
• Added a RecordPurged entity op to track a cumulative PurgedCount.

Draft — not for merge until the authoritative proto PR lands and the backend serves the RPCs on TaskHubSidecarService.

Co-authored-by: Copilot App 223556219+Copilot@users.noreply.github.com

[Copilot]

Pull request overview

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

Suppressed comments (2)

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

• DeleteExternalBlobActivity is invoked without the PurgeActivityRetryPolicy. The other purge activities use this retry policy and the PR description says activities use a small retry policy; without it, infra/worker-level activity failures will bubble up to the outer loop and delay a whole cycle instead of retrying the delete call in-place.

        BlobDeleteResult result = await context.CallActivityAsync<BlobDeleteResult>(
            nameof(DeleteExternalBlobActivity),
            tombstone.Token);

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

• This method calls the user-provided configure delegate twice (once via Services.Configure(...) and once directly via configure(probe)) to decide whether to register the hosted auto-purge starter. If the delegate has side effects (reads config with caching, increments counters, etc.) this can produce unexpected behavior. Consider avoiding the second invocation by always registering the starter and having it no-op when AutoPurge is false (or by introducing an explicit opt-in API for auto-purge).

        // Conditional DI: register the auto-purge starter only when the caller opted into auto-purge. Peek the
        // flag now by running the configure delegate against a probe (options configurators are pure setters).
        LargePayloadStorageOptions probe = new();
        configure(probe);
        if (probe.AutoPurge)
        {
            RegisterBlobPurgeJobStarter(builder);
        }

[Copilot]

Pull request overview

Copilot reviewed 28 out of 28 changed files in this pull request and generated 1 comment.

Suppressed comments (2)

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

• The configure delegate is executed immediately on a probe instance (configure(probe)) to decide whether to register the hosted purge starter. This changes the normal Options/DI semantics (the delegate will now run twice: once here and once during options resolution) and can cause unexpected side effects or early exceptions if callers do anything beyond pure property setters (e.g., read environment/config, throw on missing settings, mutate shared state). Consider registering the hosted service unconditionally and having it no-op when AutoPurge is false (or using another runtime check) so the caller-provided configure delegate is only executed by the options pipeline.

        // Conditional DI: register the auto-purge starter only when the caller opted into auto-purge. Peek the
        // flag now by running the configure delegate against a probe (options configurators are pure setters).
        LargePayloadStorageOptions probe = new();
        configure(probe);
        if (probe.AutoPurge)

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

• BlobPurgeJobStarter allocates a CancellationTokenSource but never disposes it. Disposing the CTS on stop avoids holding onto timer registrations and aligns with typical IHostedService cleanup patterns.

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        this.cts?.Cancel();

        Task? pending = this.ensureTask;

[Copilot]

Pull request overview

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

Suppressed comments (3)

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

• UseExternalizedPayloads invokes the user-provided configure delegate twice (once via Services.Configure and again on a probe instance) to detect AutoPurge. If configure has side-effects (e.g., reads env vars, performs validation, logs, throws on missing settings), those side-effects will happen twice and can be surprising.

        // Conditional DI: register the auto-purge starter only when the caller opted into auto-purge. Peek the
        // flag now by running the configure delegate against a probe (options configurators are pure setters).
        LargePayloadStorageOptions probe = new();
        configure(probe);
        if (probe.AutoPurge)

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

• BlobPurgeJob.Create stores purgeBatchSize without validating it. If an invalid value (e.g., 0 or >1000) is ever provided, the orchestrator will repeatedly fail when calling GetTombstonedPayloadsAsync (GrpcDurableTaskClient enforces 1..1000), leaving the singleton job stuck in an Active-but-broken 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:148

• DeleteExternalBlobActivity is invoked without the purge retry policy (unlike GetTombstonedPayloadsActivity/AckPurgedPayloadsActivity). This contradicts the stated design of using a small retry policy for purge activities and makes transient activity failures fail the cycle immediately instead of being retried per PurgeActivityRetryPolicy.

        BlobDeleteResult result = await context.CallActivityAsync<BlobDeleteResult>(
            nameof(DeleteExternalBlobActivity),
            tombstone.Token);

[Copilot]

Pull request overview

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

Suppressed comments (1)

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

• purgeBatchSize is stored without validating range. If the entity is created with 0/negative (e.g., via a direct entity call), the orchestrator will pass that value to GetTombstonedPayloadsAsync, which throws for limit <= 0, causing the purge job to spin/fail every cycle. Consider validating here to prevent creating a permanently-broken job state.

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

[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/Client/BlobPurgeJobStarter.cs:95

• StopAsync cancels the CTS but never awaits/observes the background ensure task (it only awaits Task.WhenAny) and never disposes the CTS. This can leave exceptions unobserved and can allow the ensure loop to keep running past shutdown if it doesn’t complete promptly. Prefer awaiting the task with a bounded cancellation token and dispose/clear the CTS/task (pattern matches other hosted services in the repo, e.g. SandboxActivityWorkerRegistrationHostedService.StopAsync).

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        this.cts?.Cancel();

        Task? pending = this.ensureTask;

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

• BlobPurgeJob.Run schedules a fixed-instance-id orchestrator without any exception handling. In this repo, entities that start orchestrations wrap ScheduleNewOrchestration in try/catch to avoid rolling back the entity operation (and losing the start action) on transient/expected errors (see src/ExportHistory/Entity/ExportJob.cs:259-293 and src/ScheduledTasks/Entity/Schedule.cs:269-306). Consider aligning with that pattern and recording LastError when scheduling fails.

        string instanceId = BlobPurgeConstants.GetOrchestratorInstanceId(context.Id.Key);
        StartOrchestrationOptions startOrchestrationOptions = new(instanceId);

        context.ScheduleNewOrchestration(
            new TaskName(nameof(BlobPurgeJobOrchestrator)),

[Copilot]

Pull request overview

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

Suppressed comments (4)

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

• BlobPurgeJob.Create persists purgeBatchSize without validating it. If a caller passes 0 or > MaxBatchSize (e.g., via the public ExecuteBlobPurgeJobOperationOrchestrator bridge), the purge orchestrator will repeatedly fail because GrpcDurableTaskClient.GetTombstonedPayloadsAsync throws for limit <= 0, resulting in an endless backoff loop with no progress. Consider validating the range here (at least for non-Active creates) so invalid values fail fast instead of permanently wedging the singleton job.

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

src/Client/Grpc/GrpcDurableTaskClient.cs:690

• AckPurgedPayloadsAsync also doesn't translate StatusCode.Unimplemented. If called against an older sidecar, it will throw RpcException instead of a consistent CLR exception (other optional RPCs in this client translate Unimplemented). Add an Unimplemented catch here too.

        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
        {
            throw new OperationCanceledException(
                $"The {nameof(this.AckPurgedPayloadsAsync)} operation was canceled.", e, cancellation);
        }

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

• This test locks in accepting purge batch size = 0, but 0 will cause GetTombstonedPayloadsAsync(limit) to throw (GrpcDurableTaskClient enforces 1..1000). If BlobPurgeJob.Create is hardened to validate its input (recommended), update this test to assert the invalid batch size is rejected.

    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.

src/Client/Grpc/GrpcDurableTaskClient.cs:648

• GetTombstonedPayloadsAsync currently only translates StatusCode.Cancelled. If the sidecar/backend is older and doesn't implement the new RPC, gRPC will return StatusCode.Unimplemented and this method will leak a raw RpcException (inconsistent with other methods in this type that translate Unimplemented to a CLR exception). Add a StatusCode.Unimplemented translation so callers get a predictable exception type.

This issue also appears on line 686 of the same file.

        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
        {
            throw new OperationCanceledException(
                $"The {nameof(this.GetTombstonedPayloadsAsync)} operation was canceled.", e, cancellation);
        }

[Copilot]

Pull request overview

Copilot reviewed 29 out of 29 changed files in this pull request and generated 1 comment.

Suppressed comments (2)

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

• BlobPurgeJob.Create persists purgeBatchSize without validating the range. The rest of the pipeline assumes 1..1000 (LargePayloadStorageOptions enforces this and GrpcDurableTaskClient.GetTombstonedPayloadsAsync throws for <=0 or >1000). Because the entity is driven via a public bridge orchestrator (ExecuteBlobPurgeJobOperationOrchestrator + BlobPurgeJobOperationRequest), external callers can still invoke Create with an invalid value, leading to a perpetual purge job that just backs off on errors. Consider validating purgeBatchSize here (or coercing to a safe default) to keep the job robust against malformed inputs.

    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;

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

• The auto-purge startup gate assumes only BlobPayloadStore can delete payloads ("if (this.store is not BlobPayloadStore)"). However PayloadStore.DeleteAsync is virtual and external implementations can override it to support deletion (including blob-token-compatible implementations). As written, such stores will be incorrectly treated as non-deletable and AutoPurge will never start. Consider checking delete capability (override) instead of concrete type.

        // Auto-purge deletes blobs through the store, but PayloadStore.DeleteAsync is virtual and its base
        // implementation throws NotSupportedException. A store that cannot delete would fail every single
        // payload, so refuse to start the job rather than spin against the backend - and rather than ack rows
        // whose blobs were never deleted, which would destroy the backend's record of what still needs cleanup.
        // This is a configuration error and is surfaced at startup, where it is cheapest to notice.
        if (this.store is not BlobPayloadStore)
        {
            this.logger.BlobPurgeStoreCannotDelete(this.store.GetType().FullName);
            return Task.CompletedTask;
        }

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