diff --git a/src/Client/Core/DurableTaskClient.cs b/src/Client/Core/DurableTaskClient.cs index 03303800..bc31915c 100644 --- a/src/Client/Core/DurableTaskClient.cs +++ b/src/Client/Core/DurableTaskClient.cs @@ -549,6 +549,30 @@ public virtual Task> ListInstanceIdsAsync( $"{this.GetType()} does not support listing orchestration instance IDs filtered by completed time."); } + /// + /// Gets a batch of tombstoned (soft-deleted) externalized payloads whose backing blobs should be deleted + /// by a credentialed caller before the backend hard-deletes the rows. + /// + /// The maximum number of tombstoned payloads to request. + /// The cancellation token. + /// The batch of tombstoned payloads whose blobs should be deleted. + /// Thrown if this implementation does not support the operation. + public virtual Task> GetTombstonedPayloadsAsync( + int limit, CancellationToken cancellation = default) + => throw new NotSupportedException($"{this.GetType()} does not support retrieving tombstoned payloads."); + + /// + /// Acknowledges tombstoned payloads whose backing blobs have been deleted so the backend can hard-delete + /// the corresponding rows. + /// + /// The payloads whose blobs have been deleted. + /// The cancellation token. + /// A task that completes when the acknowledgement has been sent. + /// Thrown if this implementation does not support the operation. + public virtual Task AckPurgedPayloadsAsync( + IEnumerable acks, CancellationToken cancellation = default) + => throw new NotSupportedException($"{this.GetType()} does not support acknowledging purged payloads."); + // TODO: Create task hub // TODO: Delete task hub diff --git a/src/Client/Core/PayloadPurgeAck.cs b/src/Client/Core/PayloadPurgeAck.cs new file mode 100644 index 00000000..6e7788aa --- /dev/null +++ b/src/Client/Core/PayloadPurgeAck.cs @@ -0,0 +1,13 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.Client; + +/// +/// Serializable acknowledgement that the worker has deleted the blob for a tombstoned payload, so the +/// backend can hard-delete the soft-deleted row. Mirrors the PayloadPurgeAck protobuf message. +/// +/// The backend partition that owns the payload row. +/// The orchestration instance key the payload belongs to. +/// The backend identifier of the soft-deleted payload row. +public sealed record PayloadPurgeAck(int PartitionId, long InstanceKey, long PayloadId); diff --git a/src/Client/Core/TombstonedPayload.cs b/src/Client/Core/TombstonedPayload.cs new file mode 100644 index 00000000..bb3f6896 --- /dev/null +++ b/src/Client/Core/TombstonedPayload.cs @@ -0,0 +1,15 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.Client; + +/// +/// Serializable representation of a tombstoned payload the backend has soft-deleted and whose blob the +/// worker should delete. Mirrors the TombstonedPayload protobuf message but is safe to pass through +/// the orchestration/activity boundary. +/// +/// The backend partition that owns the payload row. +/// The orchestration instance key the payload belongs to. +/// The backend identifier of the soft-deleted payload row. +/// The externalized payload token whose backing blob should be deleted. +public sealed record TombstonedPayload(int PartitionId, long InstanceKey, long PayloadId, string Token); diff --git a/src/Client/Grpc/GrpcDurableTaskClient.cs b/src/Client/Grpc/GrpcDurableTaskClient.cs index 23350d4c..6e79b6b2 100644 --- a/src/Client/Grpc/GrpcDurableTaskClient.cs +++ b/src/Client/Grpc/GrpcDurableTaskClient.cs @@ -624,6 +624,72 @@ public override async Task> GetOrchestrationHistoryAsync( } } + /// + public override async Task> GetTombstonedPayloadsAsync( + int limit, CancellationToken cancellation = default) + { + if (limit <= 0 || limit > 1000) + { + throw new ArgumentOutOfRangeException( + nameof(limit), limit, "Limit must be greater than 0 and less than or equal to 1000."); + } + + P.GetTombstonedPayloadsResponse response; + try + { + response = await this.sidecarClient.GetTombstonedPayloadsAsync( + new P.GetTombstonedPayloadsRequest { Limit = limit }, + cancellationToken: cancellation); + } + catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled) + { + throw new OperationCanceledException( + $"The {nameof(this.GetTombstonedPayloadsAsync)} operation was canceled.", e, cancellation); + } + + List result = new(response.Payloads.Count); + foreach (P.TombstonedPayload payload in response.Payloads) + { + result.Add(new TombstonedPayload( + payload.PartitionId, payload.InstanceKey, payload.PayloadId, payload.Token)); + } + + return result; + } + + /// + public override async Task AckPurgedPayloadsAsync( + IEnumerable acks, CancellationToken cancellation = default) + { + Check.NotNull(acks); + + P.AckPurgedPayloadsRequest request = new(); + foreach (PayloadPurgeAck ack in acks) + { + request.Acks.Add(new P.PayloadPurgeAck + { + PartitionId = ack.PartitionId, + InstanceKey = ack.InstanceKey, + PayloadId = ack.PayloadId, + }); + } + + if (request.Acks.Count == 0) + { + return; + } + + try + { + await this.sidecarClient.AckPurgedPayloadsAsync(request, cancellationToken: cancellation); + } + catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled) + { + throw new OperationCanceledException( + $"The {nameof(this.AckPurgedPayloadsAsync)} operation was canceled.", e, cancellation); + } + } + static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, ILogger logger, out CallInvoker callInvoker) { Func>? recreator = options.Internal.ChannelRecreator; diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/AckPurgedPayloadsActivity.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/AckPurgedPayloadsActivity.cs new file mode 100644 index 00000000..7ef6dbc8 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/AckPurgedPayloadsActivity.cs @@ -0,0 +1,36 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Client; +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Activity that acknowledges to the backend the payloads whose blobs the worker has deleted, so the backend +/// can hard-delete the soft-deleted rows. +/// +/// The Durable Task client used to acknowledge purged payloads to the backend. +/// The logger instance. +[DurableTask] +public class AckPurgedPayloadsActivity( + DurableTaskClient client, + ILogger logger) + : TaskActivity, object?> +{ + readonly DurableTaskClient client = Check.NotNull(client); + readonly ILogger logger = Check.NotNull(logger); + + /// + public override async Task RunAsync(TaskActivityContext context, List input) + { + if (input is null || input.Count == 0) + { + return null; + } + + await this.client.AckPurgedPayloadsAsync(input, CancellationToken.None); + this.logger.BlobPurgeAckedPayloads(input.Count); + return null; + } +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs new file mode 100644 index 00000000..12438d47 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs @@ -0,0 +1,128 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Azure; +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Activity that deletes a single externalized payload blob given its token. Deletion is idempotent, so +/// re-delivered tokens and concurrent workers are safe. +/// +/// +/// Outcome classification, verified against the Azure.Storage.Blobs / Azure.Core exception model (not +/// assumed): +/// +/// +/// The Azure SDK already retries transient failures internally (connection errors plus HTTP +/// 408/429/500/502/503/504, with exponential backoff), so any exception that escapes +/// means those built-in retries were already exhausted. +/// +/// +/// Legacy blob:v1: tokens are discarded without a delete attempt: a v1 token carries only a container +/// name, not the storage account, so a delete against the currently-configured account cannot be verified - if +/// the store has been repointed the delete would report success while the real blob survives elsewhere. The +/// backend ack protocol has no "skip" status (an un-acked row is re-served every cycle), so the token is acked +/// to keep the pipeline moving and logged at error level as the operator's recovery pointer. +/// +/// +/// Permanent failures are discarded (acked so the backend clears the row) because retrying can never succeed: +/// an from the store's token decode - a genuinely malformed or unrecognized +/// token (v1 tokens are gated out above and never reach the store); a +/// with 400 (for example +/// InvalidUri / InvalidResourceName when the decoded blob name violates Azure naming rules); and a +/// when a v2 token points at a storage account the configured credential +/// cannot reach (connection-string / account-key auth is account-specific). The backend batch is cursor-less, +/// so an undroppable row would otherwise re-stream every cycle and block the pipeline head-of-line; the +/// account-unreachable case is logged at error level so an operator can reconcile it. +/// +/// +/// Everything else is treated as transient and leaves the payload tombstoned to retry on a later cycle: +/// throttling / 5xx that outlived the SDK's retries, 403 authorization failures (which need an operator +/// credential fix rather than dropping data), timeouts / cancellation, and a +/// from a misconfigured store that cannot delete at all (retried, not dropped, so the work is recoverable once +/// a deleting store is registered). A blob is never dropped on an uncertain error, and a single bad token never +/// fails the whole batch. +/// +/// +/// +/// The payload store used to delete blobs. +/// The logger instance. +[DurableTask] +public class DeleteExternalBlobActivity( + PayloadStore store, + ILogger logger) + : TaskActivity +{ + readonly PayloadStore store = Check.NotNull(store); + readonly ILogger logger = Check.NotNull(logger); + + /// + public override async Task RunAsync(TaskActivityContext context, string input) + { + Check.NotNullOrEmpty(input, nameof(input)); + + if (input.StartsWith(BlobPayloadStore.TokenPrefixV1, StringComparison.Ordinal)) + { + // Auto-purge deliberately does not act on legacy v1 tokens. A v1 token carries only a container + // *name* - not the storage account - so a delete against the currently-configured account cannot be + // verified: if the store has since been repointed, DeleteIfExistsAsync returns false and the purge + // would silently report success while the real blob survives in the old account. Rather than delete + // on an unverifiable pointer, the token is discarded. The backend ack protocol carries no "skip" + // status (PayloadPurgeAck is just partition/instance/payload id, and an un-acked row is re-served by + // an uncursored TOP(N) query every cycle), so declining without acking would permanently block the + // pipeline. The full token is logged at error level so it remains a recoverable pointer. + this.logger.BlobPurgeDeleteV1TokenUnsupported(input); + return BlobDeleteResult.Discarded; + } + + try + { + await this.store.DeleteAsync(input, CancellationToken.None); + return BlobDeleteResult.Deleted; + } + catch (ArgumentException ex) + { + // The token is malformed or points at a different container; it can never succeed. Discard it so + // the backend clears the row instead of re-streaming the same poison token every cycle. + this.logger.BlobPurgeDeleteDiscarded(ex, input); + return BlobDeleteResult.Discarded; + } + catch (RequestFailedException ex) when (ex.Status == 400) + { + // Service rejected the request as permanently invalid (e.g. InvalidUri / InvalidResourceName - the + // decoded blob name violates Azure naming rules). Retrying can never succeed, so discard it like a + // poison token: ack so the backend clears the row instead of re-streaming it forever. + this.logger.BlobPurgeDeleteDiscarded(ex, input); + return BlobDeleteResult.Discarded; + } + catch (NotSupportedException ex) + { + // The registered store does not implement deletion (PayloadStore.DeleteAsync is virtual and its base + // implementation throws). This is a misconfiguration, not poison data: every payload would fail the + // same way, so acking would hard-delete the backend's entire record of what still needs cleanup while + // every blob survives. Keep the payload tombstoned so the work is recoverable once an operator + // registers a store that can delete. + this.logger.BlobPurgeDeleteNotSupported(ex, input); + return BlobDeleteResult.Retry; + } + catch (PayloadStorageException ex) + { + // The token is well-formed but points at a storage account this worker's credential cannot reach + // (cross-account without AAD). Retrying can never succeed from this process, and because the backend + // streams tombstones with an uncursored TOP(N) query, leaving it un-acked would re-serve the same + // token every cycle and permanently block the purge pipeline. Discard it so the row is cleared, and + // log at Error so an operator can reclaim the orphaned blob out-of-band. + this.logger.BlobPurgeDeleteUnreachable(ex, input); + return BlobDeleteResult.Discarded; + } + catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException) + { + // Transient failure: leave the payload tombstoned so a later purge cycle can retry it. A single + // bad token must not fail the whole batch. + this.logger.BlobPurgeDeleteFailed(ex, input); + return BlobDeleteResult.Retry; + } + } +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs new file mode 100644 index 00000000..838b293a --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs @@ -0,0 +1,32 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Client; +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Activity that fetches a batch of tombstoned payloads from the backend for the auto-purge job to delete. +/// +/// The Durable Task client used to query the backend for tombstoned payloads. +/// The logger instance. +[DurableTask] +public class GetTombstonedPayloadsActivity( + DurableTaskClient client, + ILogger logger) + : TaskActivity> +{ + readonly DurableTaskClient client = Check.NotNull(client); + readonly ILogger logger = Check.NotNull(logger); + + /// + public override async Task> RunAsync(TaskActivityContext context, int input) + { + int limit = input; + List payloads = + await this.client.GetTombstonedPayloadsAsync(limit, CancellationToken.None); + this.logger.BlobPurgeFetchedTombstones(payloads.Count); + return payloads; + } +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs new file mode 100644 index 00000000..b6f10d9c --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs @@ -0,0 +1,165 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using DurableTask.Core.Exceptions; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Entities; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Client-side hosted service that ensures the per-task-hub singleton blob payload auto-purge job exists. It is +/// registered unconditionally by UseExternalizedPayloads and decides what to do at startup, once options are +/// fully resolved: it no-ops silently when auto-purge is disabled, and no-ops with an error log when the +/// registered store cannot delete. It never blocks host startup - the ensure work runs on a background task +/// that retries until the backend is reachable. The job is a per-task-hub singleton, so racing client +/// processes simply no-op. +/// +sealed class BlobPurgeJobStarter : IHostedService +{ + static readonly TimeSpan RetryDelay = TimeSpan.FromSeconds(10); + + readonly IDurableTaskClientProvider clientProvider; + readonly PayloadStore store; + readonly IOptionsMonitor options; + readonly string builderName; + readonly ILogger logger; + readonly EntityInstanceId entityId = new(nameof(BlobPurgeJob), BlobPurgeConstants.JobId); + + CancellationTokenSource? cts; + Task? ensureTask; + + public BlobPurgeJobStarter( + IDurableTaskClientProvider clientProvider, + PayloadStore store, + IOptionsMonitor options, + string builderName, + ILogger logger) + { + this.clientProvider = Check.NotNull(clientProvider); + this.store = Check.NotNull(store); + this.options = Check.NotNull(options); + this.builderName = Check.NotNull(builderName); + this.logger = Check.NotNull(logger); + } + + /// + public Task StartAsync(CancellationToken cancellationToken) + { + LargePayloadStorageOptions opts = this.options.Get(this.builderName); + + // Not opted in. The starter is registered unconditionally by UseExternalizedPayloads because whether + // auto-purge is enabled can only be known once options are fully resolved - the flag can be set by the + // inline configure delegate, services.Configure, configuration binding or PostConfigure. Deciding at + // registration time (by running the delegate against a probe instance) both invoked user code twice and + // missed every enable path except the inline delegate. This is the normal path for apps that externalize + // payloads without auto-purge, so it returns silently without logging. + if (!opts.AutoPurge) + { + return Task.CompletedTask; + } + + // 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; + } + + // Resolve the client by builder name rather than by type: a named client builder must get its own + // client, and resolving lazily here - after the AutoPurge gate - avoids constructing a DurableTaskClient + // at host start for apps that externalize payloads without auto-purge. + DurableTaskClient client = this.clientProvider.GetClient(this.builderName); + + int batchSize = opts.PayloadPurgeBatchSize; + + // Do not block host startup; ensure the job on a background task with basic retry until the backend + // is reachable. + this.cts = new CancellationTokenSource(); + this.ensureTask = Task.Run(() => this.EnsureJobAsync(client, batchSize, this.cts.Token), CancellationToken.None); + return Task.CompletedTask; + } + + /// + public async Task StopAsync(CancellationToken cancellationToken) + { + this.cts?.Cancel(); + + 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); + } + } + + async Task EnsureJobAsync(DurableTaskClient client, int batchSize, CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested) + { + try + { + // The singleton is already guaranteed by the entity's fixed key (Create no-ops when the job is + // active) and the orchestrator's fixed instance id. The bridge orchestration's only job is to + // apply the entity's Create once, under a fixed instance id. Before (re)scheduling it, check the + // existing bridge: if it already Completed - or is still alive (Running/Pending/Suspended) - the + // job is set up, so do not reschedule. (Re-running a Completed bridge is wasteful: with a fixed + // id and no dedupe policy the backend would purge and replace the terminal instance on every + // host restart.) Only (re)schedule when the bridge is absent, or ended in a Failed/Terminated + // state that may never have applied Create - which lets a failed setup self-heal. + OrchestrationMetadata? existing = await client.GetInstanceAsync( + BlobPurgeConstants.StarterInstanceId, cancellationToken); + + bool needsSchedule = existing is null + or { RuntimeStatus: OrchestrationRuntimeStatus.Failed or OrchestrationRuntimeStatus.Terminated }; + if (!needsSchedule) + { + this.logger.BlobPurgeJobEnsured(); + return; + } + + BlobPurgeJobOperationRequest request = new( + this.entityId, nameof(BlobPurgeJob.Create), batchSize); + + await client.ScheduleNewOrchestrationInstanceAsync( + new TaskName(nameof(ExecuteBlobPurgeJobOperationOrchestrator)), + request, + new StartOrchestrationOptions(BlobPurgeConstants.StarterInstanceId), + cancellationToken); + + this.logger.BlobPurgeJobEnsured(); + return; + } + catch (OrchestrationAlreadyExistsException) + { + // Race: another client scheduled the bridge between our status check and schedule call. That is + // fine - the singleton is already kicked off; treat it as ensured and stop. + this.logger.BlobPurgeJobEnsured(); + return; + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + return; + } + catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException) + { + this.logger.BlobPurgeStarterRetry(ex); + try + { + await Task.Delay(RetryDelay, cancellationToken); + } + catch (OperationCanceledException) + { + return; + } + } + } + } +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs new file mode 100644 index 00000000..ca4cce32 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs @@ -0,0 +1,46 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Constants used throughout the blob payload auto-purge functionality. +/// +static class BlobPurgeConstants +{ + /// + /// The fixed, process-global job ID for the singleton blob payload auto-purge job. A single job drains + /// tombstoned payloads for the whole scheduler, so the ID is hard-coded rather than caller-supplied. + /// + public const string JobId = "__dt_blob_payload_autopurge__"; + + /// + /// The default number of tombstoned payloads the auto-purge job requests from the backend per cycle, + /// used whenever a batch size is not explicitly configured. + /// + public const int DefaultBatchSize = 500; + + /// + /// The maximum batch size the auto-purge job may request per cycle. Mirrors the gRPC + /// GetTombstonedPayloadsAsync contract, which rejects limits greater than 1000. + /// + public const int MaxBatchSize = 1000; + + /// + /// The fixed instance ID of the client-to-entity bridge orchestration the starter schedules to ensure the + /// singleton job. A fixed ID keeps racing client processes from creating duplicate bridge orchestrations. + /// + public const string StarterInstanceId = "BlobPurgeJobStarter-" + JobId; + + /// + /// The prefix used for generating blob purge job orchestrator instance IDs. Format: "BlobPurgeJob-{jobId}". + /// + public const string OrchestratorInstanceIdPrefix = "BlobPurgeJob-"; + + /// + /// Generates an orchestrator instance ID for a given blob purge job ID. + /// + /// The blob purge job ID. + /// The orchestrator instance ID. + public static string GetOrchestratorInstanceId(string jobId) => $"{OrchestratorInstanceIdPrefix}{jobId}"; +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs new file mode 100644 index 00000000..2d55a979 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs @@ -0,0 +1,84 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Entities; +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Durable entity that manages the lifecycle of the singleton blob payload auto-purge job. +/// +/// The logger instance. +class BlobPurgeJob(ILogger logger) : TaskEntity +{ + /// + /// Creates (or reactivates) the auto-purge job. Because the job is a per-task-hub singleton, this is + /// intentionally a no-op when the job is already so that extra + /// client processes racing to create it do not disturb the running job. + /// + /// The entity context. + /// + /// The maximum number of tombstoned payloads to request from the backend per cycle. + /// + 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; + + logger.BlobPurgeJobCreated(context.Id.Key); + + // Signal Run to start the perpetual purge orchestrator. + context.SignalEntity(context.Id, nameof(this.Run)); + } + + /// + /// Starts the purge orchestrator if the job is active. Uses a fixed orchestrator instance ID so only one + /// orchestrator ever runs for the singleton job. + /// + /// The entity context. + public void Run(TaskEntityContext context) + { + if (this.State.Status != BlobPurgeJobStatus.Active) + { + return; + } + + string instanceId = BlobPurgeConstants.GetOrchestratorInstanceId(context.Id.Key); + StartOrchestrationOptions startOrchestrationOptions = new(instanceId); + + context.ScheduleNewOrchestration( + new TaskName(nameof(BlobPurgeJobOrchestrator)), + new BlobPurgeJobRunRequest(context.Id, this.State.PurgeBatchSize), + startOrchestrationOptions); + + this.State.LastModifiedAt = DateTimeOffset.UtcNow; + } + + /// + /// Records progress after a purge cycle completes. + /// + /// The entity context. + /// The number of blobs purged in the cycle. + public void RecordPurged(TaskEntityContext context, long purgedCount) + { + this.State.PurgedCount += purgedCount; + this.State.LastModifiedAt = DateTimeOffset.UtcNow; + } + + /// + /// Gets the current state of the auto-purge job. + /// + /// The entity context. + /// The current job state. + public BlobPurgeJobState Get(TaskEntityContext context) => this.State; +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs new file mode 100644 index 00000000..a84a2ff8 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs @@ -0,0 +1,54 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Log messages for the Azure Blob externalized-payload auto-purge job. +/// +static partial class Logs +{ + [LoggerMessage(EventId = 810, Level = LogLevel.Information, Message = "Blob payload auto-purge job '{jobId}' created.")] + public static partial void BlobPurgeJobCreated(this ILogger logger, string? jobId); + + [LoggerMessage(EventId = 811, Level = LogLevel.Information, Message = "Blob payload auto-purge job '{jobId}' is already running; ignoring the create request.")] + public static partial void BlobPurgeJobAlreadyRunning(this ILogger logger, string? jobId); + + [LoggerMessage(EventId = 812, Level = LogLevel.Information, Message = "Blob payload auto-purge orchestrator for job '{jobId}' stopping; job status is {status}.")] + public static partial void BlobPurgeJobOrchestratorStopping(this ILogger logger, string? jobId, string status); + + [LoggerMessage(EventId = 813, Level = LogLevel.Warning, Message = "Failed to delete externalized payload blob for token '{token}'; leaving it tombstoned for a later purge cycle.")] + public static partial void BlobPurgeDeleteFailed(this ILogger logger, Exception exception, string token); + + [LoggerMessage(EventId = 814, Level = LogLevel.Debug, Message = "Blob payload auto-purge fetched {count} tombstoned payload(s) from the backend.")] + public static partial void BlobPurgeFetchedTombstones(this ILogger logger, int count); + + [LoggerMessage(EventId = 815, Level = LogLevel.Debug, Message = "Blob payload auto-purge acknowledged {count} purged payload(s) to the backend.")] + public static partial void BlobPurgeAckedPayloads(this ILogger logger, int count); + + [LoggerMessage(EventId = 817, Level = LogLevel.Information, Message = "Blob payload auto-purge singleton job ensured.")] + public static partial void BlobPurgeJobEnsured(this ILogger logger); + + [LoggerMessage(EventId = 818, Level = LogLevel.Warning, Message = "Blob payload auto-purge starter could not ensure the singleton job; retrying.")] + public static partial void BlobPurgeStarterRetry(this ILogger logger, Exception exception); + + [LoggerMessage(EventId = 819, Level = LogLevel.Warning, Message = "Discarding poison externalized payload token '{token}'; it can never be deleted, acknowledging it so the backend can clear the row.")] + public static partial void BlobPurgeDeleteDiscarded(this ILogger logger, Exception exception, string token); + + [LoggerMessage(EventId = 820, Level = LogLevel.Warning, Message = "Blob payload auto-purge cycle for job '{jobId}' failed; backing off before retrying so the job keeps running.")] + public static partial void BlobPurgeCycleFailed(this ILogger logger, Exception exception, string? jobId); + + [LoggerMessage(EventId = 821, Level = LogLevel.Error, Message = "Externalized payload token '{token}' points at a storage account the configured credential cannot reach; the blob cannot be deleted by this worker and will be orphaned. Acknowledging it so the backend can clear the row - reclaim the blob manually or reconfigure the payload store with identity (AAD) authentication that can access both accounts.")] + public static partial void BlobPurgeDeleteUnreachable(this ILogger logger, Exception exception, string token); + + [LoggerMessage(EventId = 822, Level = LogLevel.Error, Message = "Externalized payload token '{token}' uses the legacy v1 format, which auto-purge does not delete because it does not identify the storage account. The backing blob will NOT be deleted and is now orphaned; the row is acknowledged so the purge pipeline is not blocked. Reclaim the blob using the container and blob name in the token above, and upgrade to an SDK version that writes self-describing v2 tokens.")] + public static partial void BlobPurgeDeleteV1TokenUnsupported(this ILogger logger, string token); + + [LoggerMessage(EventId = 823, Level = LogLevel.Error, Message = "Blob payload auto-purge is enabled but the registered PayloadStore ('{storeType}') is not an Azure Blob payload store and cannot delete payloads. The auto-purge job was not started; externalized payloads will not be reclaimed. Register the Azure Blob payload store, or disable AutoPurge.")] + public static partial void BlobPurgeStoreCannotDelete(this ILogger logger, string? storeType); + + [LoggerMessage(EventId = 824, Level = LogLevel.Error, Message = "The registered PayloadStore does not support deleting payloads, so externalized payload token '{token}' cannot be purged. Leaving it tombstoned; register an Azure Blob payload store on the worker or disable AutoPurge.")] + public static partial void BlobPurgeDeleteNotSupported(this ILogger logger, Exception exception, string token); +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobDeleteResult.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobDeleteResult.cs new file mode 100644 index 00000000..d3386be5 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobDeleteResult.cs @@ -0,0 +1,27 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// The outcome of attempting to delete a single externalized payload blob during an auto-purge cycle. +/// +public enum BlobDeleteResult +{ + /// + /// The blob was deleted, or was already gone. The payload should be acknowledged so the backend can + /// hard-delete the row. + /// + Deleted, + + /// + /// The token is permanently invalid (poison) and can never be deleted. The payload should still be + /// acknowledged so the backend clears the stuck row instead of re-streaming the same token forever. + /// + Discarded, + + /// + /// A transient failure occurred. The payload is left tombstoned so a later purge cycle can retry it. + /// + Retry, +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobState.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobState.cs new file mode 100644 index 00000000..8bd4cdd5 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobState.cs @@ -0,0 +1,40 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// State for the singleton blob payload auto-purge job, stored in the entity. +/// +public sealed class BlobPurgeJobState +{ + /// + /// Gets or sets the current status of the auto-purge job. + /// + public BlobPurgeJobStatus Status { get; set; } + + /// + /// Gets or sets the time when the job was first created. + /// + public DateTimeOffset? CreatedAt { get; set; } + + /// + /// Gets or sets the time when the job state was last modified. + /// + public DateTimeOffset? LastModifiedAt { get; set; } + + /// + /// Gets or sets the total number of payload blobs the job has purged. + /// + public long PurgedCount { get; set; } + + /// + /// Gets or sets the last error message, if any. + /// + public string? LastError { get; set; } + + /// + /// Gets or sets the maximum number of tombstoned payloads requested from the backend per cycle. + /// + public int PurgeBatchSize { get; set; } +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobStatus.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobStatus.cs new file mode 100644 index 00000000..97c37e66 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobStatus.cs @@ -0,0 +1,21 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Represents the current status of the singleton blob payload auto-purge job. +/// +public enum BlobPurgeJobStatus +{ + /// + /// The job has not been started yet. This is the default status of a freshly initialized entity, so it is + /// kept as the zero value to avoid a brand-new entity accidentally appearing active. + /// + Pending, + + /// + /// The job is active and draining tombstoned payloads from the backend. + /// + Active, +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs new file mode 100644 index 00000000..50758e60 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs @@ -0,0 +1,167 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Entities; +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Orchestrator input describing the purge job to run. +/// +/// The entity ID of the owning . +/// The maximum number of tombstoned payloads to request per cycle. +/// The number of cycles processed since the last continue-as-new. +public sealed record BlobPurgeJobRunRequest( + EntityInstanceId JobEntityId, int PurgeBatchSize, int ProcessedCycles = 0); + +/// +/// Perpetual orchestrator that drains tombstoned payloads from the backend, deletes their blobs with capped +/// parallelism, and acknowledges the successful deletions so the backend can hard-delete the rows. It idles +/// on a timer when there is nothing to purge and continues-as-new periodically to keep its history small. +/// +[DurableTask] +public class BlobPurgeJobOrchestrator : TaskOrchestrator +{ + const int ContinueAsNewFrequency = 5; + const int MaxParallelDeletes = 32; + static readonly TimeSpan IdleDelay = TimeSpan.FromMinutes(1); + static readonly TimeSpan ErrorBackoff = TimeSpan.FromMinutes(1); + + // Retry policy for the purge activities: 3 attempts with exponential backoff (15s, 30s, capped at 60s). + static readonly RetryPolicy PurgeActivityRetryPolicy = new( + maxNumberOfAttempts: 3, + firstRetryInterval: TimeSpan.FromSeconds(15), + backoffCoefficient: 2.0, + maxRetryInterval: TimeSpan.FromSeconds(60)); + + /// + public override async Task RunAsync(TaskOrchestrationContext context, BlobPurgeJobRunRequest input) + { + ILogger logger = context.CreateReplaySafeLogger(); + string jobId = input.JobEntityId.Key; + + int batchSize = input.PurgeBatchSize; + int processedCycles = input.ProcessedCycles; + + while (true) + { + processedCycles++; + if (processedCycles > ContinueAsNewFrequency) + { + context.ContinueAsNew(new BlobPurgeJobRunRequest(input.JobEntityId, batchSize, ProcessedCycles: 0)); + return null!; + } + + try + { + // Stop cleanly if the job has been stopped or removed. + BlobPurgeJobState? state = await context.Entities.CallEntityAsync( + input.JobEntityId, nameof(BlobPurgeJob.Get), null); + + if (state is null || state.Status != BlobPurgeJobStatus.Active) + { + logger.BlobPurgeJobOrchestratorStopping(jobId, state?.Status.ToString() ?? "null"); + return null; + } + + List tombstones = await context.CallActivityAsync>( + nameof(GetTombstonedPayloadsActivity), + batchSize, + new TaskOptions(PurgeActivityRetryPolicy)); + + if (tombstones is null || tombstones.Count == 0) + { + // Nothing to purge right now: block on a timer (push-free idle) then check again. + await context.CreateTimer(IdleDelay, default); + continue; + } + + List acks = await this.DeleteBatchAsync(context, tombstones); + + if (acks.Count > 0) + { + await context.CallActivityAsync( + nameof(AckPurgedPayloadsActivity), + acks, + new TaskOptions(PurgeActivityRetryPolicy)); + + await context.Entities.CallEntityAsync( + input.JobEntityId, nameof(BlobPurgeJob.RecordPurged), (long)acks.Count); + } + else + { + // Nothing in this batch could be acknowledged: every delete returned Retry (e.g. a storage + // outage or throttling). Deletes report failure as a return value rather than an exception, + // so no retry policy or backoff applies on that path. The backend serves tombstones with an + // uncursored TOP(N) query, so continuing immediately would refetch the identical rows and + // re-attempt the identical deletes in a tight loop for as long as the outage lasts. Back off + // before trying again. + await context.CreateTimer(ErrorBackoff, default); + } + } + catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException) + { + // A single bad cycle (transient backend/entity/activity failure) must not kill the perpetual + // loop. Log, back off, then continue so the job self-heals and keeps draining. + logger.BlobPurgeCycleFailed(ex, jobId); + await context.CreateTimer(ErrorBackoff, default); + continue; + } + } + } + + async Task> DeleteBatchAsync( + TaskOrchestrationContext context, List tombstones) + { + List acks = new(tombstones.Count); + List> tasks = new(); + + foreach (TombstonedPayload tombstone in tombstones) + { + tasks.Add(this.DeleteOneAsync(context, tombstone)); + + if (tasks.Count >= MaxParallelDeletes) + { + await DrainAsync(tasks, acks); + tasks.Clear(); + } + } + + if (tasks.Count > 0) + { + await DrainAsync(tasks, acks); + } + + return acks; + } + + static async Task DrainAsync(List> tasks, List acks) + { + DeleteOutcome[] outcomes = await Task.WhenAll(tasks); + foreach (DeleteOutcome outcome in outcomes) + { + // Acknowledge blobs that were deleted (or already gone) and poison tokens that can never succeed + // so the backend can hard-delete their rows; transient failures stay tombstoned to retry. + if (outcome.ShouldAck) + { + acks.Add(outcome.Ack); + } + } + } + + async Task DeleteOneAsync(TaskOrchestrationContext context, TombstonedPayload tombstone) + { + BlobDeleteResult result = await context.CallActivityAsync( + nameof(DeleteExternalBlobActivity), + tombstone.Token, + new TaskOptions(PurgeActivityRetryPolicy)); + + return new DeleteOutcome( + result != BlobDeleteResult.Retry, + new PayloadPurgeAck(tombstone.PartitionId, tombstone.InstanceKey, tombstone.PayloadId)); + } + + readonly record struct DeleteOutcome(bool ShouldAck, PayloadPurgeAck Ack); +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/ExecuteBlobPurgeJobOperationOrchestrator.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/ExecuteBlobPurgeJobOperationOrchestrator.cs new file mode 100644 index 00000000..e8f0c66b --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/ExecuteBlobPurgeJobOperationOrchestrator.cs @@ -0,0 +1,30 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Entities; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Orchestrator that executes a single operation on a blob purge job entity and returns the result. Used as +/// a client-to-entity bridge so clients can drive the entity through the orchestration surface. +/// +[DurableTask] +public class ExecuteBlobPurgeJobOperationOrchestrator + : TaskOrchestrator +{ + /// + public override async Task RunAsync( + TaskOrchestrationContext context, BlobPurgeJobOperationRequest input) + { + return await context.Entities.CallEntityAsync(input.EntityId, input.OperationName, input.Input); + } +} + +/// +/// Request for executing a blob purge job entity operation. +/// +/// The ID of the entity to execute the operation on. +/// The name of the operation to execute. +/// Optional input for the operation. +public record BlobPurgeJobOperationRequest(EntityInstanceId EntityId, string OperationName, object? Input = null); diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs index 0817bcea..f54d499b 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs @@ -2,11 +2,15 @@ // Licensed under the MIT License. using Grpc.Core.Interceptors; +using Microsoft.DurableTask.AzureBlobPayloads; using Microsoft.DurableTask.Client; using Microsoft.DurableTask.Client.Grpc; using Microsoft.DurableTask.Converters; using Microsoft.DurableTask.Worker.Grpc.Internal; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; namespace Microsoft.DurableTask; @@ -16,6 +20,24 @@ namespace Microsoft.DurableTask; /// public static class DurableTaskClientBuilderExtensionsAzureBlobPayloads { + /// + /// Enables externalized payload storage using Azure Blob Storage for the specified client builder. + /// + /// The builder to configure. + /// The callback to configure the storage options. + /// The original builder, for call chaining. + public static IDurableTaskClientBuilder UseExternalizedPayloads( + this IDurableTaskClientBuilder builder, + Action configure) + { + Check.NotNull(builder); + Check.NotNull(configure); + + builder.Services.Configure(builder.Name, configure); + + return UseExternalizedPayloadsCore(builder); + } + /// /// Enables externalized payload storage using a pre-configured shared payload store. /// This overload helps ensure client and worker use the same configuration. @@ -31,6 +53,15 @@ public static IDurableTaskClientBuilder UseExternalizedPayloads( static IDurableTaskClientBuilder UseExternalizedPayloadsCore(IDurableTaskClientBuilder builder) { + // Reuse the shared payload store when one is already registered (e.g. via AddExternalizedPayloadStore or + // the worker builder in the same process); only register our own as a fallback so we never create a + // second, redundant PayloadStore. + builder.Services.TryAddSingleton(sp => + { + LargePayloadStorageOptions opts = sp.GetRequiredService>().Get(builder.Name); + return new BlobPayloadStore(opts); + }); + // Wrap the gRPC CallInvoker with our interceptor when using the gRPC client builder.Services .AddOptions(builder.Name) @@ -56,6 +87,23 @@ static IDurableTaskClientBuilder UseExternalizedPayloadsCore(IDurableTaskClientB } }); + // Always register the auto-purge starter. Whether auto-purge is actually enabled can only be known once + // options are fully resolved - the flag can be set by the inline configure delegate, services.Configure, + // configuration binding or PostConfigure, none of which are visible here at registration time - so the + // starter is registered unconditionally and no-ops in StartAsync when AutoPurge is disabled. + RegisterBlobPurgeJobStarter(builder); + return builder; } + + static void RegisterBlobPurgeJobStarter(IDurableTaskClientBuilder builder) + { + string builderName = builder.Name; + builder.Services.AddSingleton(sp => new BlobPurgeJobStarter( + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetRequiredService>(), + builderName, + sp.GetRequiredService>())); + } } diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs index b690d288..c63db137 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs @@ -2,11 +2,11 @@ // Licensed under the MIT License. using Grpc.Core.Interceptors; -using Grpc.Net.Client; -using Microsoft.DurableTask.Converters; +using Microsoft.DurableTask.AzureBlobPayloads; using Microsoft.DurableTask.Worker; using Microsoft.DurableTask.Worker.Grpc; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Options; using P = Microsoft.DurableTask.Protobuf; @@ -31,11 +31,6 @@ public static IDurableTaskWorkerBuilder UseExternalizedPayloads( Check.NotNull(configure); builder.Services.Configure(builder.Name, configure); - builder.Services.AddSingleton(sp => - { - LargePayloadStorageOptions opts = sp.GetRequiredService>().Get(builder.Name); - return new BlobPayloadStore(opts); - }); return UseExternalizedPayloadsCore(builder); } @@ -55,6 +50,15 @@ public static IDurableTaskWorkerBuilder UseExternalizedPayloads( static IDurableTaskWorkerBuilder UseExternalizedPayloadsCore(IDurableTaskWorkerBuilder builder) { + // Reuse the shared payload store when one is already registered (e.g. via AddExternalizedPayloadStore or + // the client builder in the same process); only register our own as a fallback so we never create a + // second, redundant PayloadStore. + builder.Services.TryAddSingleton(sp => + { + LargePayloadStorageOptions opts = sp.GetRequiredService>().Get(builder.Name); + return new BlobPayloadStore(opts); + }); + // Wrap the gRPC CallInvoker with our interceptor when using the gRPC worker builder.Services .AddOptions(builder.Name) @@ -82,6 +86,19 @@ static IDurableTaskWorkerBuilder UseExternalizedPayloadsCore(IDurableTaskWorkerB opt.Capabilities.Add(P.WorkerCapability.LargePayloads); }); + // Register the entity/orchestrators/activities that run the singleton auto-purge job. These are + // ALWAYS registered (not gated on AutoPurge) so that a client-enabled job always has something to + // execute here. The purge activities fetch/ack via the injected DurableTaskClient. + builder.AddTasks(r => + { + r.AddEntity(); + r.AddOrchestrator(); + r.AddOrchestrator(); + r.AddActivity(); + r.AddActivity(); + r.AddActivity(); + }); + return builder; } } diff --git a/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs b/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs index 6abcbdf2..9339c997 100644 --- a/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs +++ b/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs @@ -2,6 +2,7 @@ // Licensed under the MIT License. using Azure.Core; +using Microsoft.DurableTask.AzureBlobPayloads; // Intentionally no DataAnnotations to avoid extra package requirements in minimal hosts. namespace Microsoft.DurableTask; @@ -27,6 +28,7 @@ namespace Microsoft.DurableTask; public sealed class LargePayloadStorageOptions { int thresholdBytes = 256 * 1024; + int payloadPurgeBatchSize = BlobPurgeConstants.DefaultBatchSize; /// /// Initializes a new instance of the class. @@ -115,4 +117,39 @@ public int ThresholdBytes /// Defaults to true for reduced storage and bandwidth. /// public bool CompressionEnabled { get; set; } = true; + + /// + /// Gets or sets a value indicating whether the client should start the singleton blob payload auto-purge + /// job. When enabled, the job periodically drains payload rows the backend has soft-deleted and deletes + /// the corresponding blobs from customer storage (the backend has no storage credentials of its own). + /// Defaults to false (opt-in). + /// + /// + /// Auto-purge only deletes blobs referenced by self-describing blob:v2: tokens. Payloads that were + /// externalized by SDK versions that wrote legacy blob:v1: tokens are acknowledged and logged at + /// error level, but their blobs are not deleted: a v1 token does not identify the storage account, so the + /// delete cannot be verified. + /// + public bool AutoPurge { get; set; } + + /// + /// Gets or sets the maximum number of tombstoned payloads the auto-purge job requests from the backend + /// per cycle. Must be between 1 and 1000 (inclusive); values outside this range throw + /// . Defaults to 500. + /// + public int PayloadPurgeBatchSize + { + get => this.payloadPurgeBatchSize; + set + { + if (value < 1 || value > BlobPurgeConstants.MaxBatchSize) + { + throw new ArgumentOutOfRangeException( + nameof(this.PayloadPurgeBatchSize), value, + $"PayloadPurgeBatchSize must be between 1 and {BlobPurgeConstants.MaxBatchSize} (inclusive)."); + } + + this.payloadPurgeBatchSize = value; + } + } } diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs index f519b196..d67f967c 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -23,7 +23,11 @@ namespace Microsoft.DurableTask; Justification = "SemaphoreSlim does not allocate a disposable resource unless AvailableWaitHandle is accessed.")] public sealed class BlobPayloadStore : PayloadStore { - const string TokenPrefixV1 = "blob:v1:"; + /// + /// The prefix of legacy v1 payload tokens, which identify the container by name only and not the storage + /// account. Auto-purge uses this to detect and skip v1 tokens. + /// + internal const string TokenPrefixV1 = "blob:v1:"; const string TokenPrefixV2 = "blob:v2:"; const string ContentEncodingGzip = "gzip"; const int MaxRetryAttempts = 8; @@ -198,6 +202,50 @@ public override async Task DownloadAsync(string token, CancellationToken return await DownloadFromBlobAsync(blob, cancellationToken); } + /// + public override async Task DeleteAsync(string token, CancellationToken cancellationToken) + { + DecodeTokenResult decoded = DecodeToken(token); + + BlobClient blob; + if (!decoded.IsV2) + { + // v1 tokens do not carry the account, so the payload is assumed to live in the configured container. + if (!string.Equals(decoded.Container, this.containerClient.Name, StringComparison.Ordinal)) + { + throw new ArgumentException("Token container does not match configured container.", nameof(token)); + } + + blob = this.containerClient.GetBlobClient(decoded.Name); + } + else if (this.IsConfiguredContainer(decoded.ContainerUri!)) + { + // Same account and container as the configured store: reuse it (works with any auth mode). + blob = this.containerClient.GetBlobClient(decoded.Name); + } + else if (this.options.Credential != null) + { + // The payload lives in a different account (e.g. the store was repointed). Identity auth can still + // delete it as long as the credential has RBAC access to that account. + blob = new BlobClient(decoded.BlobUri, this.options.Credential, this.clientOptions); + } + else + { + throw new PayloadStorageException( + $"The externalized payload lives in a different storage account ('{decoded.ContainerUri}') than the " + + $"currently-configured payload store ('{this.containerClient.Uri}'). Cross-account payload deletes " + + "require identity (AAD) authentication with access to both accounts; connection-string / " + + "account-key credentials are account-specific and cannot delete in another account."); + } + + // Idempotent by design: DeleteIfExistsAsync returns false (rather than throwing) when the blob is + // already gone, so re-delivered tombstones and concurrent purges from multiple worker replicas are safe. + await blob.DeleteIfExistsAsync( + DeleteSnapshotsOption.IncludeSnapshots, + conditions: null, + cancellationToken: cancellationToken); + } + /// public override bool IsKnownPayloadToken(string value) { diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs index b0fe6f80..0ae5ccb6 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs @@ -24,6 +24,22 @@ public abstract class PayloadStore /// Payload string. public abstract Task DownloadAsync(string token, CancellationToken cancellationToken); + /// + /// Deletes the payload referenced by the token. Implementations that support deletion must be + /// idempotent: deleting a payload that no longer exists is a no-op and must not throw. + /// + /// + /// The default implementation throws . 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. + /// + /// The opaque reference token. + /// Cancellation token. + /// A task that completes when the payload has been deleted (or was already absent). + public virtual Task DeleteAsync(string token, CancellationToken cancellationToken) => + throw new NotSupportedException( + $"This {nameof(PayloadStore)} implementation does not support deleting payloads."); + /// /// Returns true if the specified value appears to be a token understood by this store. /// Implementations should not throw for unknown tokens. diff --git a/src/Grpc/orchestrator_service.proto b/src/Grpc/orchestrator_service.proto index 3d9194ac..af065941 100644 --- a/src/Grpc/orchestrator_service.proto +++ b/src/Grpc/orchestrator_service.proto @@ -823,6 +823,53 @@ service TaskHubSidecarService { // "Skip" graceful termination of orchestrations by immediately changing their status in storage to "terminated". // Note that a maximum of 500 orchestrations can be terminated at a time using this method. rpc SkipGracefulOrchestrationTerminations(SkipGracefulOrchestrationTerminationsRequest) returns (SkipGracefulOrchestrationTerminationsResponse); + + // Returns a batch of blob-externalized payload tombstones the backend has soft-deleted so the worker + // can delete the corresponding blobs from customer storage (the backend has no storage credentials of + // its own). The worker deletes each blob and then calls AckPurgedPayloads so the backend can + // hard-delete the soft-deleted rows. This is an opt-in, worker-driven pull model. + rpc GetTombstonedPayloads(GetTombstonedPayloadsRequest) returns (GetTombstonedPayloadsResponse); + + // Acknowledges that the worker has deleted the blobs for the identified payloads so the backend can + // hard-delete the soft-deleted rows. + rpc AckPurgedPayloads(AckPurgedPayloadsRequest) returns (AckPurgedPayloadsResponse); +} + +// Server -> client. Identifies a blob-externalized payload that the backend has soft-deleted and whose +// blob the worker should delete from customer storage. +message TombstonedPayload { + int32 partitionId = 1; + int64 instanceKey = 2; + int64 payloadId = 3; + // The externalized payload token (e.g. "blob:v1::") whose backing blob should be deleted. + string token = 4; +} + +// Client -> server. Acknowledges that the worker has deleted the blob for the identified payload so the +// backend can hard-delete the soft-deleted row. +message PayloadPurgeAck { + int32 partitionId = 1; + int64 instanceKey = 2; + int64 payloadId = 3; +} + +// Client -> server. Requests up to `limit` tombstoned payloads for the worker to delete. +message GetTombstonedPayloadsRequest { + int32 limit = 1; +} + +// Server -> client. Carries the batch of tombstoned payloads for the worker to delete. +message GetTombstonedPayloadsResponse { + repeated TombstonedPayload payloads = 1; +} + +// Client -> server. Acknowledges a batch of payloads whose blobs the worker has deleted. +message AckPurgedPayloadsRequest { + repeated PayloadPurgeAck acks = 1; +} + +// Server -> client. Empty response acknowledging the acks were recorded. +message AckPurgedPayloadsResponse { } message GetWorkItemsRequest { diff --git a/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobStarterTests.cs b/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobStarterTests.cs new file mode 100644 index 00000000..f79c79e6 --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobStarterTests.cs @@ -0,0 +1,108 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using FluentAssertions; +using Microsoft.DurableTask.AzureBlobPayloads; +using Microsoft.DurableTask.Client; +using Microsoft.Extensions.Options; +using Xunit; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests.AutoPurge; + +public class BlobPurgeJobStarterTests +{ + [Fact] + public async Task StartAsync_WhenStoreCannotDelete_DoesNotStartJobOrTouchClient() + { + // Arrange - the registered store is not the blob store, so its DeleteAsync is unsupported. Every payload + // would fail, so the starter must refuse to run rather than spin against the backend. AutoPurge is on so + // the run gets past the opt-in gate and reaches the store-capability gate under test. + Mock provider = new(); + BlobPurgeJobStarter starter = new( + provider.Object, + new NonDeletingPayloadStore(), + OptionsFor(new LargePayloadStorageOptions { AutoPurge = true }), + "test", + new TestLogger()); + + // Act + await starter.StartAsync(CancellationToken.None); + + // Assert - the store gate short-circuits before the client is resolved, so the provider is never asked + // for a client. + provider.Verify(p => p.GetClient(It.IsAny()), Times.Never); + } + + [Fact] + public async Task StartAsync_WhenStoreIsBlobStore_DoesNotShortCircuit() + { + // Arrange - the real blob store with auto-purge enabled. UseDevelopmentStorage=true is a valid connection + // string that constructs the store offline (no network I/O), so resolving it at host start is safe. + BlobPayloadStore store = new(new LargePayloadStorageOptions("UseDevelopmentStorage=true")); + Mock provider = new(); + provider.Setup(p => p.GetClient(It.IsAny())).Returns(new Mock("test").Object); + TestLogger logger = new(); + BlobPurgeJobStarter starter = new( + provider.Object, + store, + OptionsFor(new LargePayloadStorageOptions("UseDevelopmentStorage=true") { AutoPurge = true }), + "test", + logger); + + // Act + await starter.StartAsync(CancellationToken.None); + + // Assert - neither gate fired: no store-cannot-delete log, and the starter proceeded to resolve the + // client and start its background ensure path (which runs on a background task, so timing is not + // asserted). + logger.Logs.Should().NotContain(entry => entry.Message.Contains("is not an Azure Blob payload store")); + provider.Verify(p => p.GetClient(It.IsAny()), Times.Once); + + // Cleanup - cancel the background ensure task so it does not outlive the test. + await starter.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task StartAsync_WhenAutoPurgeDisabled_DoesNotResolveClientOrLog() + { + // Arrange - auto-purge is off. Even with a delete-capable blob store, the starter must no-op silently: + // it is registered unconditionally, so the not-opted-in path is the common case and must not log or + // resolve a client. + BlobPayloadStore store = new(new LargePayloadStorageOptions("UseDevelopmentStorage=true")); + Mock provider = new(); + TestLogger logger = new(); + BlobPurgeJobStarter starter = new( + provider.Object, + store, + OptionsFor(new LargePayloadStorageOptions("UseDevelopmentStorage=true") { AutoPurge = false }), + "test", + logger); + + // Act + await starter.StartAsync(CancellationToken.None); + + // Assert - returned before resolving a client and without logging anything at all. + provider.Verify(p => p.GetClient(It.IsAny()), Times.Never); + logger.Logs.Should().BeEmpty(); + } + + static IOptionsMonitor OptionsFor(LargePayloadStorageOptions options) + { + Mock> monitor = new(); + monitor.Setup(m => m.Get(It.IsAny())).Returns(options); + return monitor.Object; + } + + sealed class NonDeletingPayloadStore : PayloadStore + { + // DeleteAsync is intentionally NOT overridden: the base PayloadStore.DeleteAsync throws + // NotSupportedException, which is exactly the "store cannot delete" configuration the starter refuses. + public override Task UploadAsync(string payLoad, CancellationToken cancellationToken) => + throw new NotSupportedException(); + + public override Task DownloadAsync(string token, CancellationToken cancellationToken) => + throw new NotSupportedException(); + + public override bool IsKnownPayloadToken(string value) => false; + } +} diff --git a/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs b/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs new file mode 100644 index 00000000..2117299c --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs @@ -0,0 +1,102 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using FluentAssertions; +using Microsoft.DurableTask.AzureBlobPayloads; +using Microsoft.DurableTask.Entities.Tests; +using Xunit; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests.AutoPurge; + +public class BlobPurgeJobTests +{ + readonly BlobPurgeJob job = new(new TestLogger()); + + [Fact] + public async Task Create_WhenStopped_ActivatesJobAndStoresBatchSize() + { + // Arrange + TestEntityOperation operation = new( + nameof(BlobPurgeJob.Create), + new TestEntityState(null), + 250); + + // Act + await this.job.RunAsync(operation); + + // Assert + BlobPurgeJobState state = Assert.IsType( + operation.State.GetState(typeof(BlobPurgeJobState))); + state.Status.Should().Be(BlobPurgeJobStatus.Active); + state.PurgeBatchSize.Should().Be(250); + state.CreatedAt.Should().NotBeNull(); + state.LastModifiedAt.Should().NotBeNull(); + } + + [Fact] + public async Task Create_WhenAlreadyActive_IsNoOp() + { + // Arrange + BlobPurgeJobState existing = new() + { + Status = BlobPurgeJobStatus.Active, + PurgeBatchSize = 100, + }; + TestEntityOperation operation = new( + nameof(BlobPurgeJob.Create), + new TestEntityState(existing), + 999); + + // Act + await this.job.RunAsync(operation); + + // Assert - status stays Active and the original batch size is retained, proving the create no-op'd. + BlobPurgeJobState state = Assert.IsType( + operation.State.GetState(typeof(BlobPurgeJobState))); + state.Status.Should().Be(BlobPurgeJobStatus.Active); + state.PurgeBatchSize.Should().Be(100); + } + + [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( + operation.State.GetState(typeof(BlobPurgeJobState))); + state.PurgeBatchSize.Should().Be(0); + } + + [Fact] + public async Task Get_ReturnsCurrentState() + { + // Arrange + BlobPurgeJobState existing = new() + { + Status = BlobPurgeJobStatus.Active, + PurgeBatchSize = 42, + }; + TestEntityOperation operation = new( + nameof(BlobPurgeJob.Get), + new TestEntityState(existing), + null); + + // Act + object? result = await this.job.RunAsync(operation); + + // Assert + BlobPurgeJobState state = Assert.IsType(result); + state.Status.Should().Be(BlobPurgeJobStatus.Active); + state.PurgeBatchSize.Should().Be(42); + } +} diff --git a/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/DeleteExternalBlobActivityTests.cs b/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/DeleteExternalBlobActivityTests.cs new file mode 100644 index 00000000..93989325 --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/DeleteExternalBlobActivityTests.cs @@ -0,0 +1,122 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Azure; +using FluentAssertions; +using Microsoft.DurableTask.AzureBlobPayloads; +using Xunit; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests.AutoPurge; + +public class DeleteExternalBlobActivityTests +{ + [Fact] + public async Task RunAsync_WhenDeleteThrowsRequestFailed400_DiscardsPoisonToken() + { + // Arrange - a Status 400 (e.g. InvalidResourceName) is a permanent service rejection. + StubPayloadStore store = new(new RequestFailedException(400, "InvalidResourceName")); + DeleteExternalBlobActivity activity = new(store, new TestLogger()); + + // Act + BlobDeleteResult result = await activity.RunAsync(null!, "blob:v2:https://acct.blob.core.windows.net/payloads/bad name"); + + // Assert - discarded so the backend acks and clears the row instead of re-streaming forever. + result.Should().Be(BlobDeleteResult.Discarded); + } + + [Fact] + public async Task RunAsync_WhenDeleteThrowsRequestFailedNon400_LeavesTombstonedForRetry() + { + // Arrange - a Status 503 that escaped the SDK's internal retries is treated as transient. + StubPayloadStore store = new(new RequestFailedException(503, "ServerBusy")); + DeleteExternalBlobActivity activity = new(store, new TestLogger()); + + // Act + BlobDeleteResult result = await activity.RunAsync(null!, "blob:v2:https://acct.blob.core.windows.net/payloads/abc123"); + + // Assert - left tombstoned so a later purge cycle can retry; a blob is never dropped on doubt. + result.Should().Be(BlobDeleteResult.Retry); + } + + [Fact] + public async Task RunAsync_WhenDeleteThrowsPayloadStorageException_DiscardsToUnblockPipeline() + { + // Arrange - the payload lives in a storage account the configured credential cannot reach. Retrying can + // never succeed and the backend batch is cursor-less, so a permanently unreachable row would re-stream + // every cycle and block later rows; it must be discarded (acked), not retried. + StubPayloadStore store = new(new PayloadStorageException("cross-account delete requires identity auth")); + DeleteExternalBlobActivity activity = new(store, new TestLogger()); + + // Act + BlobDeleteResult result = await activity.RunAsync(null!, "blob:v2:https://other.blob.core.windows.net/c/abc123"); + + // Assert - discarded so the pipeline head-of-line is not blocked by an undeletable payload. + result.Should().Be(BlobDeleteResult.Discarded); + } + + [Fact] + public async Task RunAsync_V1Token_DiscardsWithoutCallingStore() + { + // Arrange - auto-purge policy: a legacy v1 token identifies no storage account, so it is dropped before + // the store is ever consulted. + Mock store = new(); + DeleteExternalBlobActivity activity = new(store.Object, new TestLogger()); + + // Act + BlobDeleteResult result = await activity.RunAsync(null!, "blob:v1:payloads:abc123"); + + // Assert - discarded by the gate, and the store's DeleteAsync was never invoked. + result.Should().Be(BlobDeleteResult.Discarded); + store.Verify(s => s.DeleteAsync(It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact] + public async Task RunAsync_V2Token_CallsStore() + { + // Arrange - a self-describing v2 token is not gated and must reach the store. + Mock store = new(); + store.Setup(s => s.DeleteAsync(It.IsAny(), It.IsAny())).Returns(Task.CompletedTask); + DeleteExternalBlobActivity activity = new(store.Object, new TestLogger()); + + // Act + BlobDeleteResult result = await activity.RunAsync(null!, "blob:v2:https://acct.blob.core.windows.net/payloads/abc123"); + + // Assert - the store deleted the blob (proves the gate is v1-only and did not break the happy path). + result.Should().Be(BlobDeleteResult.Deleted); + store.Verify(s => s.DeleteAsync(It.IsAny(), It.IsAny()), Times.Once); + } + + [Fact] + public async Task RunAsync_WhenStoreDoesNotSupportDelete_RetriesToPreserveTombstone() + { + // Arrange - a store that cannot delete (the base PayloadStore.DeleteAsync throws NotSupportedException). + StubPayloadStore store = new(new NotSupportedException()); + DeleteExternalBlobActivity activity = new(store, new TestLogger()); + + // Act + BlobDeleteResult result = await activity.RunAsync(null!, "blob:v2:https://acct.blob.core.windows.net/payloads/abc123"); + + // Assert - retried (tombstone preserved), never discarded: acking would destroy the backend's cleanup + // ledger while the blob survives. + result.Should().Be(BlobDeleteResult.Retry); + result.Should().NotBe(BlobDeleteResult.Discarded); + } + + sealed class StubPayloadStore : PayloadStore + { + readonly Exception? deleteError; + + public StubPayloadStore(Exception? deleteError) => this.deleteError = deleteError; + + public override Task DeleteAsync(string token, CancellationToken cancellationToken) => + this.deleteError is null ? Task.CompletedTask : throw this.deleteError; + + public override Task UploadAsync(string payLoad, CancellationToken cancellationToken) => + throw new NotSupportedException(); + + public override Task DownloadAsync(string token, CancellationToken cancellationToken) => + throw new NotSupportedException(); + + public override bool IsKnownPayloadToken(string value) => true; + } +} diff --git a/test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj b/test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj index 39298f69..6accf793 100644 --- a/test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj +++ b/test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj @@ -7,8 +7,13 @@ $(AssemblyName) + + + + + diff --git a/test/Extensions/AzureBlobPayloads.Tests/DependencyInjection/UseExternalizedPayloadsTests.cs b/test/Extensions/AzureBlobPayloads.Tests/DependencyInjection/UseExternalizedPayloadsTests.cs new file mode 100644 index 00000000..05013de4 --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/DependencyInjection/UseExternalizedPayloadsTests.cs @@ -0,0 +1,124 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using FluentAssertions; +using Microsoft.DurableTask.AzureBlobPayloads; +using Microsoft.DurableTask.Client; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Moq; +using Xunit; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests.DependencyInjection; + +public class UseExternalizedPayloadsTests +{ + [Fact] + public void UseExternalizedPayloads_WithAutoPurgeEnabled_RegistersHostedPurgeStarter() + { + // Arrange + ServiceCollection services = new(); + Mock builder = new(); + builder.Setup(b => b.Services).Returns(services); + builder.Setup(b => b.Name).Returns(string.Empty); + + // Act + builder.Object.UseExternalizedPayloads(options => options.AutoPurge = true); + + // Assert - the purge-job starter is the only IHostedService this path registers. + services.Should().ContainSingle(d => d.ServiceType == typeof(IHostedService)); + } + + [Fact] + public void UseExternalizedPayloads_WithAutoPurgeDisabled_StillRegistersHostedPurgeStarter() + { + // Arrange + ServiceCollection services = new(); + Mock builder = new(); + builder.Setup(b => b.Services).Returns(services); + builder.Setup(b => b.Name).Returns(string.Empty); + + // Act - auto-purge left at its default (false). + builder.Object.UseExternalizedPayloads(options => { }); + + // Assert - the starter is registered unconditionally now; whether auto-purge is enabled can only be + // known once options are fully resolved, so the no-op moved into BlobPurgeJobStarter.StartAsync. Do not + // "fix" this back to NotContain - the registration is intentional. + services.Should().ContainSingle(d => d.ServiceType == typeof(IHostedService)); + } + + [Fact] + public void UseExternalizedPayloads_AutoPurgeViaServicesConfigure_RegistersResolvableStarter() + { + // Arrange - the enable path that silently no-op'd before: AutoPurge set through services.Configure (not + // the inline delegate) plus the parameterless overload. The old probe-at-registration only saw the + // inline delegate, so the starter was never registered. It must now be registered and, more importantly, + // resolvable from the built provider. + ServiceCollection services = new(); + services.AddSingleton>(NullLogger.Instance); + services.AddSingleton(Mock.Of()); + services.Configure(o => + { + o.AutoPurge = true; + o.ConnectionString = "UseDevelopmentStorage=true"; + }); + + Mock builder = new(); + builder.Setup(b => b.Services).Returns(services); + builder.Setup(b => b.Name).Returns(string.Empty); + + // Act - the parameterless overload; AutoPurge comes from services.Configure above. + builder.Object.UseExternalizedPayloads(); + using ServiceProvider provider = services.BuildServiceProvider(); + + // Assert - the starter resolves as a hosted service (construction pulls PayloadStore, the client + // provider, options and logger), proving the whole enable path is wired end to end. + provider.GetServices().OfType().Should().ContainSingle(); + } + + [Fact] + public void UseExternalizedPayloads_ConfigureDelegate_InvokedExactlyOnce() + { + // Arrange - a delegate that counts its invocations. The old probe-at-registration ran configure a second + // time against a throwaway options instance; this locks in that user code runs exactly once, when the + // named options are first materialized. + ServiceCollection services = new(); + Mock builder = new(); + builder.Setup(b => b.Services).Returns(services); + builder.Setup(b => b.Name).Returns(string.Empty); + + int invocations = 0; + + // Act + builder.Object.UseExternalizedPayloads(options => invocations++); + using ServiceProvider provider = services.BuildServiceProvider(); + provider.GetRequiredService>().Get(string.Empty); + + // Assert - configure ran once (at options materialization), not a second time at registration. + invocations.Should().Be(1); + } + + [Fact] + public void UseExternalizedPayloads_ClientOnly_RegistersResolvablePayloadStore() + { + // Arrange - a client-only host with no worker and no explicit AddExternalizedPayloadStore. This is the + // exact shape that previously failed: the core method declared a PostConfigure dependency on + // PayloadStore without ever registering it, so options resolution threw at runtime. + ServiceCollection services = new(); + Mock builder = new(); + builder.Setup(b => b.Services).Returns(services); + builder.Setup(b => b.Name).Returns(string.Empty); + + // Act - UseDevelopmentStorage=true is a valid connection string that BlobServiceClient accepts with no + // network I/O, so the store constructs offline. Build the provider and actually resolve PayloadStore. + builder.Object.UseExternalizedPayloads(options => options.ConnectionString = "UseDevelopmentStorage=true"); + using ServiceProvider provider = services.BuildServiceProvider(); + + // Assert - the store resolves without throwing and is the blob-backed implementation. + PayloadStore store = provider.GetRequiredService(); + store.Should().BeOfType(); + } +} diff --git a/test/Extensions/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs b/test/Extensions/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs new file mode 100644 index 00000000..d608a936 --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs @@ -0,0 +1,54 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using FluentAssertions; +using Xunit; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests.Options; + +public class LargePayloadStorageOptionsTests +{ + [Fact] + public void Defaults_AutoPurgeDisabled_AndBatchSize500() + { + // Arrange & Act + LargePayloadStorageOptions options = new(); + + // Assert + options.AutoPurge.Should().BeFalse(); + options.PayloadPurgeBatchSize.Should().Be(500); + } + + [Theory] + [InlineData(0)] + [InlineData(-1)] + [InlineData(1001)] + public void PayloadPurgeBatchSize_OutOfRange_Throws(int value) + { + // Arrange + LargePayloadStorageOptions options = new(); + + // Act + Action act = () => options.PayloadPurgeBatchSize = value; + + // Assert + act.Should().Throw(); + } + + [Theory] + [InlineData(1)] + [InlineData(500)] + [InlineData(999)] + [InlineData(1000)] + public void PayloadPurgeBatchSize_InRange_IsAccepted(int value) + { + // Arrange + LargePayloadStorageOptions options = new(); + + // Act + options.PayloadPurgeBatchSize = value; + + // Assert + options.PayloadPurgeBatchSize.Should().Be(value); + } +} diff --git a/test/Extensions/AzureBlobPayloads.Tests/PayloadStore/BlobPayloadStoreDeleteTests.cs b/test/Extensions/AzureBlobPayloads.Tests/PayloadStore/BlobPayloadStoreDeleteTests.cs new file mode 100644 index 00000000..7ccab24a --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/PayloadStore/BlobPayloadStoreDeleteTests.cs @@ -0,0 +1,192 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Azure; +using Azure.Core; +using Azure.Storage.Blobs; +using Azure.Storage.Blobs.Models; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests; + +/// +/// Unit tests for , covering legacy v1 back-compatibility and the +/// self-describing v2 token resolution (same account, cross-account with identity, cross-account without). +/// +public class BlobPayloadStoreDeleteTests +{ + const string ContainerName = "payloads"; + const string ConfiguredAccountUrl = "https://myaccount.blob.core.windows.net"; + + static Mock CreateContainer(Mock blob, string expectedBlobName) + { + Mock container = new(); + container.Setup(c => c.Name).Returns(ContainerName); + container.Setup(c => c.Uri).Returns(new Uri($"{ConfiguredAccountUrl}/{ContainerName}")); + container.Setup(c => c.GetBlobClient(expectedBlobName)).Returns(blob.Object); + return container; + } + + static Mock CreateBlob(bool existed) + { + Mock blob = new(); + blob + .Setup(b => b.DeleteIfExistsAsync( + It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(Response.FromValue(existed, Mock.Of())); + return blob; + } + + [Fact] + public async Task DeleteAsync_V1Token_DeletesBackingBlobIncludingSnapshots() + { + // Arrange + Mock blob = CreateBlob(existed: true); + Mock container = CreateContainer(blob, "abc123"); + BlobPayloadStore store = new(new LargePayloadStorageOptions(), container.Object); + + // Act + await store.DeleteAsync($"blob:v1:{ContainerName}:abc123", CancellationToken.None); + + // Assert + container.Verify(c => c.GetBlobClient("abc123"), Times.Once); + blob.Verify( + b => b.DeleteIfExistsAsync(DeleteSnapshotsOption.IncludeSnapshots, null, It.IsAny()), + Times.Once); + } + + [Fact] + public async Task DeleteAsync_MissingBlob_IsIdempotentAndDoesNotThrow() + { + // Arrange + Mock blob = CreateBlob(existed: false); + Mock container = CreateContainer(blob, "missing"); + BlobPayloadStore store = new(new LargePayloadStorageOptions(), container.Object); + + // Act (a missing blob must be a no-op, not an error) + await store.DeleteAsync($"blob:v1:{ContainerName}:missing", CancellationToken.None); + + // Assert + blob.Verify( + b => b.DeleteIfExistsAsync( + It.IsAny(), It.IsAny(), It.IsAny()), + Times.Once); + } + + [Fact] + public async Task DeleteAsync_V1TokenContainerMismatch_ThrowsAndDoesNotDelete() + { + // Arrange - a v1 token does not carry the account, so its container must match the configured store. + Mock blob = CreateBlob(existed: true); + Mock container = CreateContainer(blob, "abc123"); + BlobPayloadStore store = new(new LargePayloadStorageOptions(), container.Object); + + // Act & Assert + await Assert.ThrowsAsync( + () => store.DeleteAsync("blob:v1:other-container:abc123", CancellationToken.None)); + blob.Verify( + b => b.DeleteIfExistsAsync( + It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + } + + [Theory] + [InlineData("not-a-token")] + [InlineData("blob:v1:only-container")] + [InlineData("blob:v1::blobname")] + public async Task DeleteAsync_InvalidToken_ThrowsArgumentException(string token) + { + // Arrange + Mock blob = CreateBlob(existed: true); + Mock container = CreateContainer(blob, "abc123"); + BlobPayloadStore store = new(new LargePayloadStorageOptions(), container.Object); + + // Act & Assert + await Assert.ThrowsAsync(() => store.DeleteAsync(token, CancellationToken.None)); + } + + [Fact] + public async Task DeleteAsync_V2TokenSameContainer_DeletesViaConfiguredClient() + { + // Arrange - a self-describing v2 token whose account+container match the configured store. The store + // recognizes it via IsConfiguredContainer and deletes through the existing container client (which works + // with any auth mode), never building a cross-account client. + Mock blob = CreateBlob(existed: true); + Mock container = CreateContainer(blob, "abc123"); + BlobPayloadStore store = new(new LargePayloadStorageOptions(), container.Object); + + // Act + await store.DeleteAsync($"blob:v2:{ConfiguredAccountUrl}/{ContainerName}/abc123", CancellationToken.None); + + // Assert + container.Verify(c => c.GetBlobClient("abc123"), Times.Once); + blob.Verify( + b => b.DeleteIfExistsAsync(DeleteSnapshotsOption.IncludeSnapshots, null, It.IsAny()), + Times.Once); + } + + [Fact] + public async Task DeleteAsync_V2TokenDifferentAccountWithCredential_DoesNotUseConfiguredContainer() + { + // Arrange - a v2 token pointing at a DIFFERENT account than the configured store, with identity auth + // available. The store must build a BlobClient bound to the token's own account using the credential and + // must not touch the configured container client. A credential that throws on token acquisition proves + // the cross-account path is taken without any network call (the throw short-circuits before the send). + Mock blob = CreateBlob(existed: true); + Mock container = CreateContainer(blob, "abc123"); + SentinelCredential credential = new(); + BlobPayloadStore store = new( + new LargePayloadStorageOptions(new Uri(ConfiguredAccountUrl), credential), container.Object); + string token = "blob:v2:https://otheraccount.blob.core.windows.net/othercontainer/abc123"; + + // Act + Exception error = await Assert.ThrowsAnyAsync( + () => store.DeleteAsync(token, CancellationToken.None)); + + // Assert - the cross-account BlobClient invoked our sentinel credential (directly or wrapped), proving + // that branch ran; the configured container client is never used for a different account. + Assert.True( + error is SentinelCredential.InvokedException || error.InnerException is SentinelCredential.InvokedException, + $"Expected the cross-account BlobClient to invoke the credential, but got: {error}"); + container.Verify(c => c.GetBlobClient(It.IsAny()), Times.Never); + } + + [Fact] + public async Task DeleteAsync_V2TokenDifferentAccountWithoutCredential_ThrowsPayloadStorageExceptionAndDoesNotDelete() + { + // Arrange - the configured store uses a connection string (account-key auth, no TokenCredential) and the + // token points at a different account. Account keys are account-specific, so the delete cannot cross + // accounts and must fail fast with a clear PayloadStorageException before any network call. + Mock blob = CreateBlob(existed: true); + Mock container = CreateContainer(blob, "abc123"); + BlobPayloadStore store = new(new LargePayloadStorageOptions("UseDevelopmentStorage=true"), container.Object); + string token = "blob:v2:https://otheraccount.blob.core.windows.net/othercontainer/abc123"; + + // Act + PayloadStorageException error = await Assert.ThrowsAsync( + () => store.DeleteAsync(token, CancellationToken.None)); + + // Assert - fails before touching the network or the configured container. + Assert.Contains("different storage account", error.Message, StringComparison.Ordinal); + container.Verify(c => c.GetBlobClient(It.IsAny()), Times.Never); + blob.Verify( + b => b.DeleteIfExistsAsync( + It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + } + + // A TokenCredential that throws as soon as a token is requested, proving the cross-account BlobClient path + // was taken without performing any network I/O. + sealed class SentinelCredential : TokenCredential + { + public override AccessToken GetToken(TokenRequestContext requestContext, CancellationToken cancellationToken) => + throw new InvokedException(); + + public override ValueTask GetTokenAsync( + TokenRequestContext requestContext, CancellationToken cancellationToken) => + throw new InvokedException(); + + public sealed class InvokedException : Exception + { + } + } +}