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

[berndverst]

Performance and reliability review

Requesting changes before merge.

The Azure Identity and Blob client lifetime itself is sound: BlobPayloadStore is a singleton on the working DI paths, it creates one container client/pipeline, per-blob clients reuse that pipeline and its HTTP connection pool, and Azure.Core caches and renews the access token rather than acquiring one for every delete. The gRPC client/channel is likewise long-lived when DI is configured correctly. I do not see an Entra token-acquisition or socket-lifetime leak in the normal path.

The main risks are in the purge control plane: one durable activity per blob creates excessive orchestration history/replay; all-retry batches immediately refetch the same rows; the Blob SDK retry budget can hold each 32-item wave for a long time; several supported client/worker/named-client DI topologies fail; a container mismatch is acknowledged even though the blob was not removed; and the singleton job is not self-healing once its perpetual orchestrator terminates. Mixed backend versions and Blob versioning/soft-delete semantics also need explicit handling.

Before merge, I recommend batching deletes within activities (or using Blob Batch), adding adaptive job-level backoff plus backend lease/attempt metadata, fixing the DI and named-client resolution paths, separating malformed-token errors from configuration mismatches, reconciling the actual perpetual orchestration, and adding provider-resolution and full-cycle tests.

References: credential reuse, Azure SDK client lifetime, Blob retry guidance, delete semantics, and versioning.

[berndverst]

[High / performance] This creates one durable activity per blob. At the 1,000-item limit, a cycle emits 1,000 activity scheduled/completed pairs; five full cycles before ContinueAsNew produce at least ~10,000 history events and roughly 160 replay waves. The durable control-plane cost can dominate the DELETE calls. Please move deletion into chunked activities (for example 25-100 tokens each), or use Blob Batch where appropriate, and parallelize within the activity.

[berndverst]

[High / reliability] If every delete returns Retry, acks.Count is zero, no exception is raised, and this loop immediately fetches the same non-empty batch again. A persistent 403, lease/configuration failure, unsupported store, or storage outage can therefore create a hot loop and keep the oldest rows blocking later work. Add exponential backoff with jitter for zero/high-failure batches, and consider cursor/lease/attempt metadata in the backend contract so failed rows do not cause head-of-line blocking.

[berndverst]

[High / performance] Bounded concurrency is good, but each of these 32 activities invokes the existing Blob pipeline configured with eight retries and a two-minute network timeout. A timeout path can occupy a wave for roughly 18 minutes before the orchestrator advances; processing 1,000 rows can consequently take many hours, followed by an immediate refetch if none succeeded. For bulk purge, use a smaller per-operation SDK retry budget and let the job-level policy provide adaptive backoff/circuit breaking. Fast persistent 403s generally bypass SDK retries and hit the hot-loop path even sooner.

[berndverst]

[High / DI] This configured client overload never registers PayloadStore, while UseExternalizedPayloadsCore adds a PostConfigure<PayloadStore, ...> dependency. A client-only host using this overload fails when the client/options are resolved unless an undocumented worker/shared registration happens to provide the store. Please register the store here or make AddExternalizedPayloadStore an explicit, validated prerequisite.

[berndverst]

[High / DI] GetTombstonedPayloadsActivity and AckPurgedPayloadsActivity both require DurableTaskClient, but AddDurableTaskWorker does not register one. In a supported split client/worker deployment, activity activation fails at dispatch. Please either perform fetch/ack through the worker's existing backend channel or explicitly register/require a client for this topology.

[berndverst]

[High / lifecycle] This checks only the short-lived bridge orchestration. Once that bridge is Completed, the starter returns even if the actual fixed-ID BlobPurgeJobOrchestrator later fails or is terminated; the entity remains Active, so Create also no-ops and the job never recovers. Please reconcile the perpetual instance itself and define restart/stop/update behavior so disabling auto-purge or changing batch size is not permanently ignored.

[berndverst]

[Medium / configuration] Invoking the caller's configure callback a second time assumes it is a pure setter; it may construct a credential or have other side effects. More importantly, this probe is the only place the starter is registered, so shared configuration via AddExternalizedPayloadStore(... AutoPurge = true) plus the no-argument overload silently ignores auto-purge. Register the hosted service once and have it inspect the resolved named options in StartAsync.

[berndverst]

[Medium / compatibility] During a mixed rollout, an older backend returns gRPC Unimplemented for this new RPC. Only cancellation is translated here, so the activity/orchestrator retries the unsupported operation indefinitely without a clear terminal diagnostic. Please add capability negotiation or map Unimplemented to an explicit unsupported-backend state that stops/disables the purge job.

[berndverst]

[Medium / storage semantics] IncludeSnapshots does not delete blob versions. With versioning enabled, deleting the base blob turns the current version into a retained previous version, so acknowledging this as purged may reclaim no storage; soft delete likewise retains bytes until expiration. Please either require/document an appropriate lifecycle policy, explicitly enumerate/delete versions, or make the user-visible status clear that this only deletes the current base blob.

[berndverst]

[Medium / tests] Descriptor presence does not exercise the runtime dependency graph, which is why the missing client-side PayloadStore, worker-only DurableTaskClient, and named-client resolution failures are not detected. Please build the provider and resolve the actual named/default client, worker task factories, options, and hosted service for client-only, worker-only, combined, and named configurations. The purge orchestrator/starter also need full-cycle tests for zero acknowledgements, replay/history growth, restart behavior, and backend Unimplemented.

[Copilot]

Pull request overview

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

Comments suppressed due to low confidence (4)

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

• This invokes the caller-provided configure delegate twice (once via Services.Configure(...) and again against a probe instance). If the delegate has side-effects or is non-deterministic, this can lead to incorrect AutoPurge detection or unexpected behavior. Consider registering the hosted service unconditionally and having it no-op when AutoPurge is false (or otherwise avoid double-invoking the delegate).

        // 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/PayloadStore/BlobPayloadStore.cs:176

• This call mixes a named argument (conditions:) with a later positional argument (cancellationToken), which is not valid C# and will fail to compile. The cancellation token should be passed as a named argument as well.

        await blob.DeleteIfExistsAsync(
            DeleteSnapshotsOption.IncludeSnapshots,
            conditions: null,
            cancellationToken);

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

• Task.WhenAny(...) returns without observing exceptions from ensureTask when it faults. This can lead to unobserved task exceptions (and can surface via TaskScheduler.UnobservedTaskException). If the task completes, it should be awaited (inside a try/catch) to observe any fault/cancellation.

        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/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs:35

• Adding a new virtual member is generally binary compatible, but it can still be a behavioral/source-breaking change for existing external subclasses that already defined a same-signature DeleteAsync method (it may become hidden, and calls via a PayloadStore reference will bind to the new base implementation). The remarks currently state this is non-breaking; it would be safer to document this edge case explicitly.

    /// The default implementation throws <see cref="NotSupportedException"/>. Stores that externalize
    /// payloads to deletable storage (for example Azure Blob Storage) should override it. It is declared
    /// virtual rather than abstract so that adding it does not break existing external subclasses.
    /// </remarks>

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