Add large-payload blob auto-purge (worker/SDK side)

[YunchuWang]

Summary

Implements the worker/SDK side of large-payload blob auto-purge in the existing Microsoft.DurableTask.Extensions.AzureBlobPayloads package (shipped as a new version — no new package).

Large orchestration payloads are externalized to Azure Blob Storage as tokens blob:v1:<container>:<blobName>. The DTS backend stores those tokens but cannot delete the blobs — it has no storage credentials. New design: the backend soft-deletes blob-externalized payload rows and streams their tokens to the worker over a bidirectional gRPC stream; the worker deletes the blobs immediately and acks; the backend then hard-deletes the row. This PR implements the worker side.

There is no customer-facing retention config, no options, and no separate enable API — the backend owns retention and decides what to purge; the SDK just drains whatever the backend streams. On the OSS worker everything is gated on the existing UseExternalizedPayloads(...) call.

Changes

• PayloadStore.DeleteAsync(token, ct) — new virtual method on the payload-store abstraction, with a default that throws NotSupportedException so existing external subclasses keep compiling (non-breaking). BlobPayloadStore overrides it: decodes the blob:v1: token to container + blob name and calls DeleteIfExistsAsync (idempotent: deleting a missing blob is a no-op, so redelivered tombstones and concurrent replicas are safe).
• BlobPayloadAutoPurgeService (public sealed BackgroundService) — opens one bidirectional gRPC stream (PurgeExternalPayloads). Push model: blocks awaiting server pushes when idle (not a polling loop). For each pushed tombstone it deletes the blob then acks on the same stream. A failed delete for one token is logged and skipped without acking — so the backend keeps the row soft-deleted and can re-push it — and the stream keeps draining (one bad token can't kill it). Resilient reconnect with jittered backoff (1s→30s cap) on stream drop/error; stops cleanly on host shutdown via the stopping token.
• Reusable across repos — the service is public and independently constructable from a CallInvoker (or gRPC ChannelBase) + a PayloadStore + an optional ILogger (defaults to NullLogger), so hosts that enable large payloads through a different code path (e.g. the DTS Azure Managed worker SDK, which wires payloads via its own AzureBlobPayloadCallInvokerFactory rather than UseExternalizedPayloads) can construct it directly and start/stop it via IHostedService. No dependency on the OSS DI path.
• Auto-registration (OSS path) — for UseExternalizedPayloads(...) the service is registered via an internal constructor that resolves the same CallInvoker the worker already uses to reach DTS, lazily at start. If the enable API is never called, the service is never registered and never runs.
• Proto — added PurgeExternalPayloads rpc plus TombstonedPayload (server→client) and PayloadPurgeAck (client→server ack) messages to TaskHubSidecarService in src/Grpc/orchestrator_service.proto (the service the worker already dials). Existing field numbers are preserved; field names use camelCase to match the rest of that file.
• Tests — focused unit tests for BlobPayloadStore.DeleteAsync (token decode, correct blob targeted, idempotency, container-mismatch / invalid-token validation) and for BlobPayloadAutoPurgeService construction from a CallInvoker/ChannelBase + PayloadStore (locks in the public reuse contract, plain-ILogger acceptance + null-arg guards), in a new test/AzureBlobPayloads.Tests project.

gRPC contract

The SDK/worker is the gRPC client dialing DTS on the worker-facing TaskHubSidecarService (wire path /TaskHubSidecarService/PurgeExternalPayloads):

rpc PurgeExternalPayloads(stream PayloadPurgeAck) returns (stream TombstonedPayload);
message TombstonedPayload { int32 partitionId = 1; int64 instanceKey = 2; int64 payloadId = 3; string token = 4; } // server -> client
message PayloadPurgeAck   { int32 partitionId = 1; int64 instanceKey = 2; int64 payloadId = 3; }                 // client -> server ack

Placement on TaskHubSidecarService is confirmed correct: the durabletask-dotnet worker (including Worker.AzureManaged) speaks only this service. Field names use camelCase to match orchestrator_service.proto convention; field numbers/types are wire-compatible with the backend. The ack message is named PayloadPurgeAck (an acknowledgement sent upstream by the worker) to fit the file's naming and de-confuse the bidi signature.

Follow-up: the authoritative proto change lives in a separate PR in microsoft/durabletask-protobuf (#76 — adds PurgeExternalPayloads + these two messages to TaskHubSidecarService, with the same PayloadPurgeAck name). This PR vendors the rpc/messages into src/Grpc/orchestrator_service.proto (where the SDK generates its gRPC stubs at build time) so the draft builds and tests standalone. Once the upstream PR merges, this file should be re-synced from upstream (content will be identical). The backend is being reconciled to serve this rpc on the worker-facing TaskHubSidecarService surface.

Notes

• The OSS DI path reuses the worker's intercepted CallInvoker (guaranteed set after the UseExternalizedPayloads post-configure). The payload interceptor only wraps unary/server-streaming calls, so the bidi purge stream passes through untouched.
• PayloadStore.DeleteAsync is virtual (not abstract) — no breaking change for external PayloadStore subclasses.
• TFMs unchanged (netstandard2.0;net6.0;net8.0;net10.0); the service avoids APIs unavailable on netstandard2.0.

Verification

• dotnet build Microsoft.DurableTask.sln — succeeds (0 errors).
• dotnet test test/AzureBlobPayloads.Tests — 11/11 pass.

Draft — kept as a draft while the backend PR lands.