diff --git a/src/Client/Core/DurableTaskClient.cs b/src/Client/Core/DurableTaskClient.cs index 03303800..c903574a 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 due large-payload tombstones whose backing blobs a credentialed caller must delete. + /// + /// The maximum number of tombstones to request. + /// The cancellation token. + /// The batch of tombstones whose blobs should be deleted. + /// Thrown if this implementation does not support the operation. + public virtual Task> GetLargePayloadTombstonesAsync( + int limit, CancellationToken cancellation = default) + => throw new NotSupportedException($"{this.GetType()} does not support retrieving large-payload tombstones."); + + /// + /// Reports the outcome of each attempted large-payload blob deletion so the backend can resolve, + /// reschedule, or quarantine the corresponding tombstones. Every attempted row is reported, not just the + /// successful ones; the backend owns retry scheduling. + /// + /// The per-row outcomes of the attempted deletions. + /// The cancellation token. + /// A task that completes when the outcomes have been recorded. + /// Thrown if this implementation does not support the operation. + public virtual Task ReportLargePayloadPurgeResultsAsync( + IEnumerable results, CancellationToken cancellation = default) + => throw new NotSupportedException($"{this.GetType()} does not support reporting large-payload purge results."); + // TODO: Create task hub // TODO: Delete task hub diff --git a/src/Client/Core/LargePayloadPurgeDisposition.cs b/src/Client/Core/LargePayloadPurgeDisposition.cs new file mode 100644 index 00000000..bfdedca2 --- /dev/null +++ b/src/Client/Core/LargePayloadPurgeDisposition.cs @@ -0,0 +1,33 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.Client; + +/// +/// The outcome of a single large-payload blob deletion attempt. The split is by whether a failure can +/// self-heal. Mirrors the LargePayloadPurgeDisposition protobuf enum. +/// +public enum LargePayloadPurgeDisposition +{ + /// + /// No disposition was specified. + /// + Unspecified = 0, + + /// + /// Terminal success. The blob was deleted, was already absent, or was deliberately left in place because + /// it is not owned by the payload store. The backend deletes the tombstone in all three cases. + /// + Deleted = 1, + + /// + /// The failure may self-heal, so the row stays pending and the backend sets the next attempt. + /// + Retry = 2, + + /// + /// A deterministic failure or protocol violation that retrying can never fix. The backend preserves the + /// evidence, alerts, and stops automatic retries. + /// + Quarantined = 3, +} diff --git a/src/Client/Core/LargePayloadPurgeResult.cs b/src/Client/Core/LargePayloadPurgeResult.cs new file mode 100644 index 00000000..0a3ee98a --- /dev/null +++ b/src/Client/Core/LargePayloadPurgeResult.cs @@ -0,0 +1,35 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.Client; + +/// +/// Serializable outcome of exactly one attempted large-payload blob deletion. Mirrors the +/// LargePayloadPurgeResult protobuf message but is safe to pass through the orchestration/activity +/// boundary. The backend owns retry scheduling and branches solely on +/// : it deletes rows reported as +/// , reschedules +/// on its own backoff, and moves +/// rows out of the active fetch. The worker never +/// computes a retry delay. +/// +/// +/// The disposition is deliberately the only outcome field: anything finer would be write-only on the backend. +/// Why an attempt failed stays in the worker's own telemetry, which holds the cause at full fidelity rather +/// than as a lossy classification, and a row is correlated to it by +/// (, , ). +/// +/// The backend partition that owns the tombstoned row. +/// The orchestration instance key the payload belonged to. +/// The backend identifier of the tombstoned payload row. +/// +/// The revision echoed unmodified from the fetched ; used by the backend +/// as a compare-and-swap guard. +/// +/// The disposition of the deletion attempt. +public sealed record LargePayloadPurgeResult( + int PartitionId, + long InstanceKey, + long PayloadId, + long Revision, + LargePayloadPurgeDisposition Disposition); diff --git a/src/Client/Core/LargePayloadTombstone.cs b/src/Client/Core/LargePayloadTombstone.cs new file mode 100644 index 00000000..c11a40db --- /dev/null +++ b/src/Client/Core/LargePayloadTombstone.cs @@ -0,0 +1,23 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.Client; + +/// +/// Serializable representation of a tombstoned large-payload row whose external blob a credentialed caller +/// must delete. Mirrors the LargePayloadTombstone protobuf message but is safe to pass through the +/// orchestration/activity boundary. +/// +/// The backend partition that owns the tombstoned row. +/// The orchestration instance key the payload belonged to. +/// The backend identifier of the tombstoned payload row. +/// +/// The self-describing blob:v2:{fullBlobUrl} payload token whose backing blob should be deleted. +/// +/// +/// An optimistic-concurrency guard echoed back unmodified in the corresponding +/// so the backend can reject duplicate or stale reports without taking +/// a per-row lease. +/// +public sealed record LargePayloadTombstone( + int PartitionId, long InstanceKey, long PayloadId, string Token, long Revision); diff --git a/src/Client/Grpc/GrpcDurableTaskClient.cs b/src/Client/Grpc/GrpcDurableTaskClient.cs index 23350d4c..0fdbf161 100644 --- a/src/Client/Grpc/GrpcDurableTaskClient.cs +++ b/src/Client/Grpc/GrpcDurableTaskClient.cs @@ -624,6 +624,82 @@ public override async Task> GetOrchestrationHistoryAsync( } } + /// + public override async Task> GetLargePayloadTombstonesAsync( + 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.GetLargePayloadTombstonesResponse response; + try + { + response = await this.sidecarClient.GetLargePayloadTombstonesAsync( + new P.GetLargePayloadTombstonesRequest { Limit = limit }, + cancellationToken: cancellation); + } + catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled) + { + throw new OperationCanceledException( + $"The {nameof(this.GetLargePayloadTombstonesAsync)} operation was canceled.", e, cancellation); + } + + List result = new(response.Tombstones.Count); + foreach (P.LargePayloadTombstone tombstone in response.Tombstones) + { + result.Add(new LargePayloadTombstone( + tombstone.PartitionId, + tombstone.InstanceKey, + tombstone.PayloadId, + tombstone.Token, + tombstone.Revision)); + } + + return result; + } + + /// + public override async Task ReportLargePayloadPurgeResultsAsync( + IEnumerable results, CancellationToken cancellation = default) + { + Check.NotNull(results); + + P.ReportLargePayloadPurgeResultsRequest request = new(); + foreach (LargePayloadPurgeResult result in results) + { + request.Results.Add(new P.LargePayloadPurgeResult + { + PartitionId = result.PartitionId, + InstanceKey = result.InstanceKey, + PayloadId = result.PayloadId, + Revision = result.Revision, + + // The managed disposition enum declares the same numeric values as its protobuf counterpart, + // so it maps across by value. This is the only enum on the message and it only travels + // outbound, so the SDK can never receive a value it does not know. + Disposition = (P.LargePayloadPurgeDisposition)result.Disposition, + }); + } + + if (request.Results.Count == 0) + { + return; + } + + try + { + await this.sidecarClient.ReportLargePayloadPurgeResultsAsync(request, cancellationToken: cancellation); + } + catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled) + { + throw new OperationCanceledException( + $"The {nameof(this.ReportLargePayloadPurgeResultsAsync)} 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/DeleteExternalBlobActivity.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs new file mode 100644 index 00000000..4d74b601 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs @@ -0,0 +1,182 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Globalization; +using System.Net; +using Azure; +using Microsoft.DurableTask.Client; +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Activity that deletes a single externalized payload blob given its token, and classifies the attempt as +/// , , or +/// . Deletion is idempotent, so re-delivered tokens and +/// concurrent workers are safe. +/// +/// +/// The split between retry and quarantine is whether the failure can self-heal, 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. It is still +/// classified as retryable, because the backend - not this activity - owns retry scheduling and can defer the +/// row past a storage outage. +/// +/// +/// Quarantine is reserved for deterministic failures and protocol violations that retrying can never fix: a +/// known version prefix whose body does not parse, a request storage rejected as permanently invalid, and a +/// legacy v1 token. Quarantine preserves the row and its token as durable evidence, so a permanent failure +/// neither blocks the queue nor destroys the only record of the blob. +/// +/// +/// A blob is never deleted on an uncertain error, and a single bad token never fails the whole batch: a +/// failure is returned as a disposition rather than thrown. +/// +/// +/// The reported result carries the disposition alone, so every branch below logs its cause where the cause is +/// still exact, rather than deriving it afterwards from a value that crossed the wire. That log is the only +/// record of why an attempt failed. Per design ยง7 it still carries neither the token nor raw exception text - +/// a token exposes the storage account, container, and blob path - so the cause is a bounded classification +/// string plus a bounded, sanitized storage error code. The token itself is preserved on the backend's +/// quarantined row. +/// +/// 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)); + + return await this.DeleteAsync(input); + } + + /// + /// Extracts a bounded, sanitized storage error code for diagnostics. The service's own error code (for + /// example BlobNotFound) is a fixed vocabulary and the numeric status is the fallback, so neither + /// can carry a token or raw exception text. + /// + static string SanitizeErrorCode(RequestFailedException exception) + { + // Pattern-matched rather than string.IsNullOrEmpty: on netstandard2.0 that method carries no + // [NotNullWhen(false)] annotation, so flow analysis cannot prove the else branch is non-null and warns. + // A constant pattern is analyzed by the compiler itself and so behaves the same on every target. + string? errorCode = exception.ErrorCode; + return errorCode is null or "" + ? exception.Status.ToString(CultureInfo.InvariantCulture) + : errorCode; + } + + async Task DeleteAsync(string token) + { + // Classify the token's version prefix before consulting the store. The store reports every token it + // cannot decode as the same ArgumentException, but the three cases have opposite dispositions, so they + // are separated here, where the prefix is still visible. + if (token.StartsWith(BlobPayloadStore.TokenPrefixV1, StringComparison.Ordinal)) + { + // A v1 token carries a container *name* but not the storage account, so a delete against the + // currently-configured account cannot be verified: if the store has since been repointed, + // DeleteIfExists returns false and the purge would falsely report success while the real blob + // survives in the old account. Retrying cannot fix that, and a success-shaped discard would destroy + // the only durable record of the blob, so the row is quarantined instead - the backend preserves + // its token as evidence and stops polling it. The backend excludes v1 at insertion time, so + // reaching this branch is an invariant violation rather than an expected path. + this.logger.BlobPurgeDeleteQuarantined("LegacyV1Token", null); + return new BlobPurgeOutcome(LargePayloadPurgeDisposition.Quarantined); + } + + if (!token.StartsWith(BlobPayloadStore.TokenPrefixV2, StringComparison.Ordinal)) + { + // An unrecognized prefix is most likely a token written by a newer SDK than this worker runs. That + // recovers after an upgrade, so it earns a deferral rather than quarantine. Quarantine is + // permanent and requires an operator to unwind; a deferral only leaves the row idle and visible, + // so an unrecognized token is deliberately kept on the recoverable side of that asymmetry. + this.logger.BlobPurgeDeleteRetryable("UnsupportedTokenVersion", null); + return new BlobPurgeOutcome(LargePayloadPurgeDisposition.Retry); + } + + try + { + PayloadDeleteOutcome outcome = await this.store.DeleteAsync(token, CancellationToken.None); + + // The blob exists but this store never wrote it, so it was left untouched. That is an expected + // outcome, not a defect - the token text merely matched the v2 grammar - and quarantining it would + // fill the quarantine set with non-defects. The tombstone is still resolved, because a blob the + // store does not own is not the store's to delete. + if (outcome == PayloadDeleteOutcome.NotStoreOwned) + { + this.logger.BlobPurgeBlobNotStoreOwned(); + } + + // Deleted, AlreadyAbsent, and NotStoreOwned are all terminal successes: none can be improved by + // trying again. + return new BlobPurgeOutcome(LargePayloadPurgeDisposition.Deleted); + } + catch (ArgumentException) + { + // The prefix gate above proves this is a v2 token, so the only remaining decode failure is a v2 + // body that does not parse. The SDK and backend control both sides of the protocol, so that + // indicates a producer, corruption, or compatibility bug; retrying can never fix it. + this.logger.BlobPurgeDeleteQuarantined("MalformedToken", null); + return new BlobPurgeOutcome(LargePayloadPurgeDisposition.Quarantined); + } + catch (NotSupportedException) + { + // The registered store does not implement deletion. Every payload would fail the same way, so the + // work is kept recoverable until an operator registers a store that can delete. + this.logger.BlobPurgeDeleteRetryable("StoreCannotDelete", null); + return new BlobPurgeOutcome(LargePayloadPurgeDisposition.Retry); + } + catch (PayloadStorageException) + { + // The token is well formed but points at a storage account this worker's credential cannot reach + // (account-key auth is account-specific). Recoverable after a configuration or credential change, + // so it is deferred rather than discarded. + this.logger.BlobPurgeDeleteRetryable("StorageAccountUnreachable", null); + return new BlobPurgeOutcome(LargePayloadPurgeDisposition.Retry); + } + catch (RequestFailedException ex) when (ex.Status == (int)HttpStatusCode.BadRequest) + { + // Storage rejected a request generated from a well-formed token as permanently invalid (for + // example InvalidUri / InvalidResourceName). Retrying can never succeed. + this.logger.BlobPurgeDeleteQuarantined("InvalidStorageRequest", SanitizeErrorCode(ex)); + return new BlobPurgeOutcome(LargePayloadPurgeDisposition.Quarantined); + } + catch (RequestFailedException ex) when ( + ex.Status == (int)HttpStatusCode.Unauthorized || ex.Status == (int)HttpStatusCode.Forbidden) + { + // Authorization can be transient or fixed by reconfiguration, so it stays recoverable rather than + // dropping data an operator can still reclaim. + this.logger.BlobPurgeDeleteRetryable("StorageAuthorizationFailed", SanitizeErrorCode(ex)); + return new BlobPurgeOutcome(LargePayloadPurgeDisposition.Retry); + } + catch (RequestFailedException ex) + { + // Throttling, 5xx, and anything else the service reported, including a failed If-Match on the + // ownership check: transient by default. + this.logger.BlobPurgeDeleteRetryable("TransientStorageFailure", SanitizeErrorCode(ex)); + return new BlobPurgeOutcome(LargePayloadPurgeDisposition.Retry); + } + catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException) + { + // Timeouts, cancellation, and network failures. A blob is never dropped on an uncertain error. + // Storage reported no code here, so the exception's type name is appended to the cause: it is a + // bounded value that cannot carry a token, and it is the only thing separating a timeout from a + // cancellation or a DNS failure now that no classification crosses the wire. + this.logger.BlobPurgeDeleteRetryable($"UnexpectedFailure:{ex.GetType().Name}", null); + return new BlobPurgeOutcome(LargePayloadPurgeDisposition.Retry); + } + } +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetLargePayloadTombstonesActivity.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetLargePayloadTombstonesActivity.cs new file mode 100644 index 00000000..273989ea --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetLargePayloadTombstonesActivity.cs @@ -0,0 +1,33 @@ +// 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 bounded batch of due large-payload tombstones from the backend for the auto-purge +/// job to delete. +/// +/// The Durable Task client used to query the backend for tombstones. +/// The logger instance. +[DurableTask] +public class GetLargePayloadTombstonesActivity( + 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 tombstones = + await this.client.GetLargePayloadTombstonesAsync(limit, CancellationToken.None); + this.logger.BlobPurgeFetchedTombstones(tombstones.Count); + return tombstones; + } +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/ReportLargePayloadPurgeResultsActivity.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/ReportLargePayloadPurgeResultsActivity.cs new file mode 100644 index 00000000..7003b7c9 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/ReportLargePayloadPurgeResultsActivity.cs @@ -0,0 +1,39 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Client; +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Activity that reports the outcome of every attempted blob deletion to the backend, so it can delete the +/// resolved tombstones, reschedule the retryable ones, and quarantine the rest. Every attempted row is +/// reported, not only the successful ones: the backend owns retry scheduling, so a row it hears nothing about +/// would simply be re-served unchanged on the next cycle. +/// +/// The Durable Task client used to report purge results to the backend. +/// The logger instance. +[DurableTask] +public class ReportLargePayloadPurgeResultsActivity( + 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.ReportLargePayloadPurgeResultsAsync(input, CancellationToken.None); + this.logger.BlobPurgeReportedResults(input.Count); + return null; + } +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs new file mode 100644 index 00000000..af197a99 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs @@ -0,0 +1,373 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using DurableTask.Core.Exceptions; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.Entities; +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 reconciles the per-task-hub singleton blob payload auto-purge job with the +/// resolved options. It is registered unconditionally by UseExternalizedPayloads and decides what to do at +/// startup, once options are fully resolved: it ensures the job exists when auto-purge is enabled, stops a +/// running job when it is disabled, and no-ops with an error log when the registered store cannot delete. It +/// never blocks host startup - the work runs on a background task that retries until the backend is reachable +/// and then, on the enabled path, keeps reconciling on a fixed interval for the lifetime of the process. +/// The job is a per-task-hub singleton, so racing client processes converge on the same result. +/// +sealed class BlobPurgeJobStarter : IHostedService, IDisposable +{ + 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? backgroundTask; + + /// + /// Initializes a new instance of the class. + /// + /// The provider used to resolve the named durable task client. + /// The registered payload store, checked for delete support before starting the job. + /// The monitor used to read the fully-resolved large payload storage options. + /// The name of the client builder this starter belongs to. + /// The logger. + 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); + } + + /// + /// Gets or sets the interval between reconciliation passes on the enabled path. Settable only so tests can + /// drive the loop without waiting on wall-clock time; nothing in the product ever assigns it. + /// + /// + /// + /// This value is the worst-case time a dead job stays dead. Nothing else re-signals the job's orchestrator, + /// so recovery cannot be faster than one pass. What it buys that back with is one schedule call per host + /// per interval, forever - so it trades a bounded recovery time against a standing, fleet-sized cost, and + /// shortening it makes that cost grow in proportion. + /// + /// + /// That cost is per host, not per task hub. Hosts start at different times, so their intervals are + /// staggered, and a pass almost always finds the previous bridge already Completed - which this call leaves + /// replaceable on purpose - so it schedules and runs a bridge of its own. The + /// path only absorbs passes that overlap a bridge which is + /// still Pending or Running, and that window is narrow because a bridge makes one entity call and exits. So + /// a fleet of N hosts costs roughly N bridge runs and N entity calls per interval, not one. + /// + /// + /// What keeps the actual purge work single is not that dedupe policy but the fixed orchestrator instance id: + /// every bridge run signals Run, and the backend discards a start aimed at an orchestrator that is still + /// alive. One perpetual orchestrator therefore serves the whole fleet however many bridges ran. + /// + /// + internal TimeSpan ReconcileInterval { get; set; } = TimeSpan.FromMinutes(5); + + /// + 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. + if (!opts.AutoPurge) + { + // Turning auto-purge off has to reach the backend to mean anything. The job is a perpetual + // orchestrator owned by the task hub, not by this process, so a job created while the flag was on + // keeps deleting blobs forever no matter how many hosts start with it off. Stop a running one. + // + // The store gate below deliberately does NOT apply to this path. Stopping a job requires no ability + // to delete anything, and a user who has switched to a store that cannot delete is precisely + // someone who needs the job stopped; that gate's error is worded for the enabled case and would be + // a false alarm here. + // + // The job's state is read before anything is signalled, so an app that never enabled auto-purge + // signals nothing and no entity is created for it. + this.cts = new CancellationTokenSource(); + this.backgroundTask = Task.Run(() => this.SignalJobStopAsync(this.cts.Token), CancellationToken.None); + 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, and keep reconciling it from there. + this.cts = new CancellationTokenSource(); + this.backgroundTask = 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.backgroundTask; + if (pending is not null) + { + // The background loop observes cancellation and returns promptly; swallow any faulted/cancelled + // result. + await Task.WhenAny(pending, Task.Delay(Timeout.Infinite, cancellationToken)).ConfigureAwait(false); + } + } + + /// + /// Disposes the cancellation source backing the background task. + /// + /// + /// Deliberately not disposed in : that method stops waiting as soon as the host's + /// shutdown token fires, so the background task may still hold this source's token. Disposing it there would + /// fault that still-running task with an when it next registers a + /// callback. The container disposes singletons after every has returned, which is + /// the safe point. + /// + public void Dispose() + { + this.cts?.Dispose(); + } + + async Task EnsureJobAsync(DurableTaskClient client, int batchSize, CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested) + { + TimeSpan delay; + + try + { + BlobPurgeJobOperationRequest request = new( + this.entityId, nameof(BlobPurgeJob.Create), batchSize); + + // The bridge orchestration's only job is to apply the entity's Create once, under a fixed + // instance id. Whether it should be (re)scheduled is decided by the backend as part of the + // create call, not by reading its status first: a read-then-schedule pair is two independent + // RPCs with no atomicity between them, so two hosts starting together can both observe "absent" + // and both schedule. + // + // The dedupe list is an inverted whitelist. The wire policy is computed as + // (all statuses - dedupe statuses) = the statuses that may be REPLACED, so any status left out + // of this list silently becomes replaceable. Deduping exactly Pending and Running therefore + // means: while the bridge is alive leave it alone, and in any other state re-run it. + // + // Pending is the subtle one and is not optional. Instances are created Pending and only become + // Running after awaiting their first task, so omitting it would leave a real window in which a + // just-scheduled bridge is replaced. + // + // Re-running a bridge that already finished is safe and close to free: the bridge's only effect + // is calling Create, which no-ops while the entity is Active, so the cost is one instance + // replacement plus one entity call per reconciliation pass. + // + // Re-running is also what lets the job self-heal after the entity is removed, for example by + // CleanEntityStorageAsync. The perpetual orchestrator exits cleanly when it reads back a null + // entity state, and a removed entity is back to its default Pending status, so the job is left + // dead with a Completed bridge behind it. Keeping Completed deduped would keep it dead + // permanently; making it replaceable means the next pass re-runs Create, which finds the + // entity not Active and rebuilds the job. + // + // This also recovers the case where the perpetual orchestrator dies while the entity is still + // Active. Create leaves the entity's status alone in that case, but it re-signals Run, and the + // resulting start is resolved by the backend: discarded outright while the orchestrator is + // alive, and allowed to replace it once it has completed, terminated, failed or been canceled. + // So a healthy job is left untouched and a dead one is rebuilt, without this side ever having to + // ask which of the two it is looking at. + // + // Recovery repeats on a fixed interval rather than being bounded by host starts. This loop runs + // for the lifetime of the process and re-issues the same call every ReconcileInterval, so an + // orchestrator that dies mid-lifetime is rebuilt within roughly one interval instead of staying + // down until the next deployment. Repeating it costs nothing extra to reason about, because a + // pass that finds everything healthy is already a no-op at both hops: the bridge's Create + // no-ops while the entity is Active, and the Run it signals is discarded by the backend while + // the orchestrator is alive. + await client.ScheduleNewOrchestrationInstanceAsync( + new TaskName(nameof(ExecuteBlobPurgeJobOperationOrchestrator)), + request, + new StartOrchestrationOptions(BlobPurgeConstants.StarterInstanceId) + .WithDedupeStatuses( + OrchestrationRuntimeStatus.Pending, + OrchestrationRuntimeStatus.Running), + cancellationToken); + + this.logger.BlobPurgeJobEnsured(); + delay = this.ReconcileInterval; + } + catch (OrchestrationAlreadyExistsException) + { + // Thrown only when the bridge already exists in one of the dedupe statuses above, so under this + // policy it means another host scheduled the bridge and it is still Pending or Running. That is + // exactly the concurrent-start race this replaced a status check to close: one create wins and + // the loser lands here. Either way the singleton is already being set up, so treat it as + // ensured and wait for the next pass. + // + // Note this is NOT the steady-state path. A bridge that already finished is Completed, which is + // replaceable, so a later pass re-runs it rather than landing here. + this.logger.BlobPurgeJobEnsured(); + delay = this.ReconcileInterval; + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + return; + } + catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException) + { + // Deliberately the short delay and not the reconcile interval. A backend that is unreachable + // when the host starts has to be retried quickly, because until one pass succeeds the job may + // not exist at all; the long interval is only the price of keeping a job that already exists + // healthy. + this.logger.BlobPurgeStarterRetry(ex); + delay = RetryDelay; + } + + try + { + await Task.Delay(delay, cancellationToken); + } + catch (OperationCanceledException) + { + // Shutdown. This is the only thing that ends the loop on the success path, which is why + // StopAsync can cancel a pass that is parked here instead of waiting out the interval. + return; + } + } + } + + async Task SignalJobStopAsync(CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested) + { + try + { + // Resolved inside the loop rather than on the host-start path. This path is taken by every app + // that externalizes payloads without auto-purge, and constructing a DurableTaskClient at host + // start for them is exactly what the enable path's lazy resolution avoids. It also means a + // client that cannot be resolved yet is retried here instead of throwing out of host startup, + // which must never happen for a feature the user has turned off. + DurableTaskClient client = this.clientProvider.GetClient(this.builderName); + + // Held in a local so the query and the signal below are issued against the same object. Both go + // through this one property, which is also where every "entities are not supported" gate in this + // SDK lives, so a client that cannot answer the query cannot receive the signal either. + DurableEntityClient entities = client.Entities; + + // Defaults to true so that every path which fails to establish the job's state falls through to + // signalling. Signalling is the correctness-critical outcome - a job left running keeps deleting + // payloads - while skipping it is only an optimization, so the optimization must never be the + // reason the stop does not happen. + bool stopNeeded = true; + + try + { + // A query against the instance store, not an entity operation: it never dispatches to the + // entity, so it cannot materialize one. That is the whole point of reading through the + // client here instead of relying on the entity's own idempotence. Signalling Stop + // unconditionally would create the job entity - with default, never-started state - for + // every app that externalizes payloads without auto-purge, and would persist a fresh write + // on every host start for an app whose job is already stopped. + // The three-argument overload is called deliberately. The shorter GetEntityAsync(id, + // cancellation) is a virtual forwarder that supplies includeState itself, and binding to a + // forwarder rather than to the abstract method is what made an earlier set of tests in this + // feature pass against a mock that was never wired up. Naming includeState also states the + // requirement directly: this call exists to read the status, so metadata alone is useless. + EntityMetadata? metadata = await entities + .GetEntityAsync(this.entityId, includeState: true, cancellationToken); + + // IncludesState is checked rather than reading metadata.State directly: State throws when + // the metadata carries none, and an entity whose state has been cleared still reports as + // existing until entity storage is cleaned. Such an entity has no running job, so it is + // treated the same as an absent one. + BlobPurgeJobState? state = metadata is { IncludesState: true } ? metadata.State : null; + + stopNeeded = state?.Status == BlobPurgeJobStatus.Active; + + if (!stopNeeded) + { + this.logger.BlobPurgeJobNotRunning(); + } + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + return; + } + catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException) + { + // Deliberately not rethrown into the retry loop. Retrying would be right for a transient + // failure but fatal for a permanent one: a client that never answers this query would keep + // the loop spinning and the running job would never be told to stop. Falling through costs + // at most the entity write this pre-check exists to avoid. + this.logger.BlobPurgeJobStateUnknown(ex); + } + + if (!stopNeeded) + { + return; + } + + // Signalled rather than driven through the bridge orchestration: unlike Create, Stop needs no + // fixed-instance dedupe. It is idempotent in the entity, so concurrent hosts converge on the + // same state and a repeat costs nothing beyond the signal itself. + await entities.SignalEntityAsync( + this.entityId, + nameof(BlobPurgeJob.Stop), + null, + null, + cancellationToken); + + this.logger.BlobPurgeJobStopRequested(); + 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..0138da35 --- /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 + /// GetLargePayloadTombstones 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..f1b704e7 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs @@ -0,0 +1,176 @@ +// 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 the auto-purge job, and starts its orchestrator if one is not already running. Because the job is + /// a per-task-hub singleton, client processes racing to create it converge on the same result rather than + /// disturbing a running job. It also takes the batch size when the job is already + /// , which is what lets a configuration change reach a 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) + { + // The batch size is taken because this is the only path by which a changed configuration reaches an + // active job - the orchestrator re-reads it from here every cycle. Without this the value written by + // the very first Create would be the only one the job ever used, and a batch size the backend + // rejects would wedge it permanently. + // + // Written only when the value actually differs, which is what keeps LastModifiedAt tracking real + // configuration changes. Create runs on every reconciliation pass, not only at host start, so an + // unconditional write would reduce the field to "time of the last pass". An entity written by a + // build that predates this field carries zero, which differs from any configured size, so the + // first Create after an upgrade still repairs it. + if (this.State.PurgeBatchSize != purgeBatchSize) + { + this.State.PurgeBatchSize = purgeBatchSize; + this.State.LastModifiedAt = DateTimeOffset.UtcNow; + } + + logger.BlobPurgeJobAlreadyRunning(context.Id.Key); + + // Run is re-signalled even though the job is already active, and this is what makes the job + // self-heal: Create runs on every reconciliation pass, so a job whose orchestrator has died is + // rebuilt within roughly one interval rather than waiting for the next host start. The signal is + // deliberately blind. An entity-initiated start carries no reuse policy, so + // the backend decides its fate: it discards the start while the target instance exists in any + // non-completed status, and purges and replaces it once the instance has completed, terminated, + // failed or been canceled. A healthy orchestrator is therefore left strictly alone and only a dead + // one is replaced. + // + // Being blind is the point, not a shortcut. The backend reaches that decision atomically, so + // delegating it removes the race entirely. Checking the orchestrator's status here and signalling + // only when it looked dead would reintroduce the window in which it dies - or recovers - between + // the read and the signal, which is strictly worse than not asking. + context.SignalEntity(context.Id, nameof(this.Run)); + 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. + /// + /// + /// + /// The orchestrator runs under a fixed instance ID, which is what keeps the singleton a singleton. No reuse + /// policy is passed, and none can be: the entity's start action has no field to carry one, so anything set + /// here would be dropped before it reached the wire. That default is the behaviour the job relies on rather + /// than an omission - the backend discards a start aimed at an instance that already exists in a + /// non-completed status, and replaces the instance only once it has completed, terminated, failed or been + /// canceled. Signalling this operation is therefore always safe, whatever the orchestrator is doing. + /// + /// + /// This operation deliberately writes no state. It schedules an orchestrator and nothing more, and it runs + /// on every reconciliation pass, so touching here would + /// overwrite a real change with the time of a pass that changed nothing. + /// + /// + /// 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); + } + + /// + /// Stops the auto-purge job. + /// + /// + /// + /// The perpetual orchestrator is deliberately not terminated here, and this operation does not touch its + /// instance ID at all. The orchestrator reads this entity at the top of every cycle and exits on its own + /// once the job is no longer , so shutdown is cooperative: there is + /// no window in which one party terminates an orchestrator that the other believes is healthy, and the + /// in-flight cycle finishes rather than being cut off part-way through a batch of deletes. + /// + /// + /// , and + /// are preserved. They are the job's history and its + /// configuration, both of which are wanted if it is started again, and keeping CreatedAt is also what + /// distinguishes a stopped job from one that was never started. + /// + /// + /// The entity context. + public void Stop(TaskEntityContext context) + { + if (this.State.Status != BlobPurgeJobStatus.Active) + { + // Load-bearing, and NOT redundant with the client-side pre-check in BlobPurgeJobStarter that + // normally prevents this operation from being signalled at all. That pre-check reads the job state + // and then signals: two separate round trips with nothing holding the state still between them, so + // the job can stop - or never have existed - in the gap. This guard is what makes losing that race + // harmless, which is the only reason the pre-check is allowed to be a plain read. Deleting it + // because "the caller already checked" would reintroduce exactly the window it was written to + // absorb. Concurrent hosts signalling at once land here for the same reason. + // + // Returning here also leaves the state exactly as it was found, which is what keeps + // LastModifiedAt meaning "when this job stopped" rather than "when a stop was last signalled at + // it". That only holds because no other operation rewrites the field on a no-op either: Run never + // writes it, and Create rewrites it only when the batch size actually differs. Breaking either of + // those breaks this too. It does not avoid materializing the entity - the framework persists + // entity state after every operation, so a stop signal to an entity that does not exist yet + // creates it with default state. Not creating it is the pre-check's job, not this guard's. + logger.BlobPurgeJobAlreadyStopped(context.Id.Key); + return; + } + + this.State.Status = BlobPurgeJobStatus.Pending; + this.State.LastModifiedAt = DateTimeOffset.UtcNow; + + logger.BlobPurgeJobStopped(context.Id.Key); + } + + /// + /// 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..3f3ae217 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs @@ -0,0 +1,63 @@ +// 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.Debug, Message = "Blob payload auto-purge job '{jobId}' is already active. Its batch size was reconciled against the current configuration, and its orchestrator was re-signalled, which starts one only if none is running.")] + 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 = "Blob payload auto-purge quarantined a payload; cause '{cause}', storage code '{storageCode}'. The failure is deterministic and cannot succeed on a retry. The backend preserves the tombstone row and its token as evidence and stops polling it. The reported result carries the disposition alone, so this log is the only record of the cause.")] + public static partial void BlobPurgeDeleteQuarantined(this ILogger logger, string cause, string? storageCode); + + [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 reported {count} purge result(s) to the backend.")] + public static partial void BlobPurgeReportedResults(this ILogger logger, int count); + + [LoggerMessage(EventId = 816, Level = LogLevel.Information, Message = "Blob payload auto-purge job '{jobId}' stopped. The perpetual orchestrator is not terminated; it reads the job state at the start of its next cycle and exits on its own.")] + public static partial void BlobPurgeJobStopped(this ILogger logger, string? jobId); + + [LoggerMessage(EventId = 817, Level = LogLevel.Debug, 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 reach the singleton job; retrying.")] + public static partial void BlobPurgeStarterRetry(this ILogger logger, Exception exception); + + [LoggerMessage(EventId = 819, Level = LogLevel.Warning, Message = "Blob payload auto-purge could not delete a payload; cause '{cause}', storage code '{storageCode}'. The backend reschedules the tombstone for a later attempt. The reported result carries the disposition alone, so this log is the only record of the cause.")] + public static partial void BlobPurgeDeleteRetryable(this ILogger logger, string cause, string? storageCode); + + [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.Warning, Message = "An externalized payload blob does not carry this store's ownership marker, so it was left untouched; the tombstone is still resolved. This is expected for payloads written before the marker shipped, and for blobs the store never created whose token text matches the payload token grammar.")] + public static partial void BlobPurgeBlobNotStoreOwned(this ILogger logger); + + [LoggerMessage(EventId = 822, Level = LogLevel.Debug, Message = "Blob payload auto-purge job '{jobId}' is already stopped; ignoring the stop request.")] + public static partial void BlobPurgeJobAlreadyStopped(this ILogger logger, string? jobId); + + [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.Information, Message = "Blob payload auto-purge is disabled, so a stop was requested for the singleton job. A job left running by an earlier configuration exits after its current cycle.")] + public static partial void BlobPurgeJobStopRequested(this ILogger logger); + + [LoggerMessage(EventId = 825, Level = LogLevel.Debug, Message = "Blob payload auto-purge is disabled and no running singleton job was found, so no stop was requested.")] + public static partial void BlobPurgeJobNotRunning(this ILogger logger); + + [LoggerMessage(EventId = 826, Level = LogLevel.Warning, Message = "Blob payload auto-purge is disabled but the singleton job's state could not be read; requesting a stop anyway. Stopping a job that may be deleting payloads matters more than avoiding a redundant stop request, which the job ignores.")] + public static partial void BlobPurgeJobStateUnknown(this ILogger logger, Exception exception); +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobState.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobState.cs new file mode 100644 index 00000000..0a17caae --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobState.cs @@ -0,0 +1,48 @@ +// 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 of the last meaningful change to the job: when it was started, when it was + /// stopped, when it was given a different from the one it already had, or + /// when it last recorded a non-zero number of purged blobs. + /// + /// + /// This is not a liveness or heartbeat signal, and it must not be read as one. Neither starting a host nor + /// a reconciliation pass moves it, and an active job whose cycles keep finding nothing to purge leaves it + /// untouched indefinitely, so a value far in the past is equally consistent with a healthy idle job and a + /// dead one. + /// + 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..fe6fe99c --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobStatus.cs @@ -0,0 +1,23 @@ +// 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 is not running. This is both the state of a job that has never been started and the resting + /// state of one that has been stopped, which distinguishes: it is + /// null only for a job that was never created. It is kept as the zero value so a brand-new entity does not + /// accidentally appear active. + /// + Pending, + + /// + /// The job is active and draining tombstoned payloads from the backend. + /// + Active, +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeOutcome.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeOutcome.cs new file mode 100644 index 00000000..7f4a1714 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeOutcome.cs @@ -0,0 +1,19 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Client; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// The outcome of attempting to delete a single externalized payload blob during an auto-purge cycle. The +/// orchestrator combines it with the tombstone's identity and revision to build the reported +/// . +/// +/// +/// Carries the disposition alone, because that is the only outcome field the contract reports. Why an attempt +/// reached its disposition is logged by at the point it is +/// classified, at higher fidelity than any value that could be carried here. +/// +/// Whether the row is resolved, should be retried, or must be quarantined. +public sealed record BlobPurgeOutcome(LargePayloadPurgeDisposition Disposition); diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs new file mode 100644 index 00000000..a4e455e4 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs @@ -0,0 +1,207 @@ +// 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 due large-payload tombstones from the backend, deletes their blobs with +/// capped parallelism, and reports every outcome so the backend can resolve, reschedule, or quarantine each +/// row. 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. BlobPurgeJob.Stop is what makes this + // reachable: it moves the entity off Active without touching this orchestrator at all, so + // shutdown is cooperative - the in-flight cycle finishes and the loop exits on its own terms + // rather than being terminated part-way through a batch of deletes. + // input: null is named deliberately. A bare positional null binds to the (id, name, options) + // overload instead, which reads as if an input were being passed when it is not. + BlobPurgeJobState? state = await context.Entities.CallEntityAsync( + input.JobEntityId, nameof(BlobPurgeJob.Get), input: null); + + if (state is null || state.Status != BlobPurgeJobStatus.Active) + { + logger.BlobPurgeJobOrchestratorStopping(jobId, state?.Status.ToString() ?? "null"); + return null; + } + + // Take the batch size from the entity rather than from this orchestrator's input. A perpetual + // orchestrator outlives configuration changes: its input is fixed when it is created and is + // carried verbatim through every continue-as-new, so using input.PurgeBatchSize would pin the + // value written by the very first Create for the entire life of the job. Re-reading it from the + // state fetch this cycle already performs costs no extra call and is what lets a changed batch + // size actually take effect. That matters because a batch size the backend rejects fails every + // fetch: without this the job would be wedged with no recovery short of deleting the entity. + // + // Fall back to the input when the stored value is not positive. An entity written by an older + // build carries no batch size at all, and asking the backend for zero rows every cycle would be + // a silent, permanent stall. + int cycleBatchSize = state.PurgeBatchSize > 0 ? state.PurgeBatchSize : batchSize; + + List tombstones = await context.CallActivityAsync>( + nameof(GetLargePayloadTombstonesActivity), + cycleBatchSize, + 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 results = await this.DeleteBatchAsync(context, tombstones); + + // Every attempted row produces a result, including the retryable ones: the backend owns retry + // scheduling, so it needs to hear about a failure to defer the row. Reporting unconditionally + // is what keeps a failing row from being re-served unchanged on the very next cycle. + await context.CallActivityAsync( + nameof(ReportLargePayloadPurgeResultsActivity), + results, + new TaskOptions(PurgeActivityRetryPolicy)); + + // Two different questions, deliberately not conflated. Progress counts only payloads that were + // actually purged; the backoff decision asks whether ANY row left the retry queue, because a + // quarantined row also stops being re-served even though nothing was reclaimed. + int purged = CountDisposition(results, LargePayloadPurgeDisposition.Deleted); + int resolved = results.Count - CountDisposition(results, LargePayloadPurgeDisposition.Retry); + + if (purged > 0) + { + await context.Entities.CallEntityAsync( + input.JobEntityId, nameof(BlobPurgeJob.RecordPurged), (long)purged); + } + + if (resolved == 0) + { + // The whole batch came back retryable (e.g. a storage outage or throttling). Deletes report + // failure as a return value rather than an exception, so no activity retry policy applies on + // that path. Continuing immediately would refetch and re-attempt in a tight loop for as long + // as the outage lasts, so back off before the next cycle. + 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; + } + } + } + + static int CountDisposition( + List results, LargePayloadPurgeDisposition disposition) + { + int count = 0; + foreach (LargePayloadPurgeResult result in results) + { + if (result.Disposition == disposition) + { + count++; + } + } + + return count; + } + + static async Task DrainAsync( + List> tasks, List results) + { + LargePayloadPurgeResult[] completed = await Task.WhenAll(tasks); + results.AddRange(completed); + } + + async Task> DeleteBatchAsync( + TaskOrchestrationContext context, List tombstones) + { + List results = new(tombstones.Count); + List> tasks = new(); + + foreach (LargePayloadTombstone tombstone in tombstones) + { + tasks.Add(this.DeleteOneAsync(context, tombstone)); + + if (tasks.Count >= MaxParallelDeletes) + { + await DrainAsync(tasks, results); + tasks.Clear(); + } + } + + if (tasks.Count > 0) + { + await DrainAsync(tasks, results); + } + + return results; + } + + async Task DeleteOneAsync( + TaskOrchestrationContext context, LargePayloadTombstone tombstone) + { + BlobPurgeOutcome outcome = await context.CallActivityAsync( + nameof(DeleteExternalBlobActivity), + tombstone.Token, + new TaskOptions(PurgeActivityRetryPolicy)); + + // The revision is echoed back unchanged so the backend can detect a tombstone that was rewritten while + // this attempt was in flight and ignore the stale result. Retry scheduling is the backend's job, so no + // next-attempt time is computed here. + return new LargePayloadPurgeResult( + tombstone.PartitionId, + tombstone.InstanceKey, + tombstone.PayloadId, + tombstone.Revision, + outcome.Disposition); + } +} 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..142f96eb 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) @@ -80,8 +84,26 @@ static IDurableTaskWorkerBuilder UseExternalizedPayloadsCore(IDurableTaskWorkerB } opt.Capabilities.Add(P.WorkerCapability.LargePayloads); + + // Item 2 of the pre-merge checklist: the resolved AutoPurge value rides the work-item + // handshake so the backend only tombstones payloads for a task hub whose workers opted in. + // Absent means "no opinion"; an explicit true/false is the customer's choice. + opt.LargePayloadAutoPurgeEnabled = opts.AutoPurge; }); + // 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/report 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..49cb020d 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,42 @@ 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 reclaims only blobs referenced by self-describing blob:v2: tokens. Payloads written by + /// SDK versions that emitted legacy blob:v1: tokens are not reclaimed, because a v1 token identifies + /// the container by name only and not the storage account, so the delete cannot be verified. Their backing + /// blobs remain in storage exactly as they did before auto-purge existed - this is not a new leak introduced + /// by auto-purge - and the backend removes their rows normally. Upgrading to an SDK version that writes v2 + /// tokens makes newly-externalized payloads eligible. + /// + 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..bc597e4c 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -23,8 +23,33 @@ 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:"; - const string TokenPrefixV2 = "blob:v2:"; + /// + /// 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:"; + + /// + /// The prefix of self-describing v2 payload tokens, which carry the blob's absolute URI including the + /// storage account. Auto-purge uses this to tell a malformed v2 token (a protocol defect) apart from an + /// unrecognized version prefix (a token written by a newer SDK). + /// + internal const string TokenPrefixV2 = "blob:v2:"; + + /// + /// The metadata name of the ownership marker written on every blob this store creates. Recognizing a + /// token proves only that its text matches the store's grammar; the marker is what proves the store + /// actually wrote the blob, so a customer's own blob is never deleted just because an orchestration + /// referenced it. Azure requires metadata names to follow the naming rules for C# identifiers, so the + /// marker is spelled with an underscore rather than a hyphen. + /// + internal const string OwnershipMarkerName = "managed_by"; + + /// + /// The fixed value of the ownership marker written on every blob this store creates. + /// + internal const string OwnershipMarkerValue = "dts"; + const string ContentEncodingGzip = "gzip"; const int MaxRetryAttempts = 8; const int BaseDelayMs = 250; @@ -120,6 +145,7 @@ public override async Task UploadAsync(string payLoad, CancellationToken BlobOpenWriteOptions writeOptions = new() { HttpHeaders = new BlobHttpHeaders { ContentEncoding = ContentEncodingGzip }, + Metadata = CreateOwnershipMetadata(), }; using Stream blobStream = await blob.OpenWriteAsync(true, writeOptions, cancellationToken); using GZipStream compressedBlobStream = new(blobStream, System.IO.Compression.CompressionLevel.Optimal, leaveOpen: true); @@ -133,7 +159,15 @@ public override async Task UploadAsync(string payLoad, CancellationToken } else { - using Stream blobStream = await blob.OpenWriteAsync(true, default, cancellationToken); + // The uncompressed path still needs write options purely to carry the ownership marker: + // the marker must be written by every path that creates a blob, or auto-purge would later + // decline to delete the store's own uncompressed payloads. It rides along in the PUT the + // upload already issues, so it costs no extra request. + BlobOpenWriteOptions writeOptions = new() + { + Metadata = CreateOwnershipMetadata(), + }; + using Stream blobStream = await blob.OpenWriteAsync(true, writeOptions, cancellationToken); // using MemoryStream payloadStream = new(payloadBuffer, writable: false); // await payloadStream.CopyToAsync(blobStream, bufferSize: DefaultCopyBufferSize, cancellationToken); @@ -198,6 +232,81 @@ 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."); + } + + // Recognizing the token proves only that its text matches this store's grammar - not that this store + // wrote the blob. A customer may keep an expensive dataset in Blob Storage and have orchestrations + // reference it by URL; deleting that would destroy data the store never created. So ownership is read + // from the object itself before anything is deleted. + BlobProperties properties; + try + { + Response response = await blob.GetPropertiesAsync( + conditions: null, cancellationToken: cancellationToken); + properties = response.Value; + } + catch (RequestFailedException ex) when (ex.Status == (int)HttpStatusCode.NotFound) + { + // Already gone. Deletion is idempotent, so a re-delivered tombstone or a concurrent worker + // replica that won the race is a success, not an error. + return PayloadDeleteOutcome.AlreadyAbsent; + } + + if (!HasOwnershipMarker(properties.Metadata)) + { + // Positive evidence that the blob is customer-owned: leave it untouched. The caller still resolves + // the payload reference, because a blob this store never wrote is not this store's to delete. + return PayloadDeleteOutcome.NotStoreOwned; + } + + // Pair the ownership read with the delete using the ETag from that same read. If anything rewrites the + // blob in between - including a customer overwriting it with content that no longer carries the marker + // - the If-Match condition fails the delete instead of removing the newer content, so the read-then- + // delete behaves as a single check-and-delete without taking a lease. + // 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. + Response deleted = await blob.DeleteIfExistsAsync( + DeleteSnapshotsOption.IncludeSnapshots, + conditions: new BlobRequestConditions { IfMatch = properties.ETag }, + cancellationToken: cancellationToken); + + return deleted.Value ? PayloadDeleteOutcome.Deleted : PayloadDeleteOutcome.AlreadyAbsent; + } + /// public override bool IsKnownPayloadToken(string value) { @@ -261,6 +370,38 @@ internal static DecodeTokenResult DecodeToken(string token) throw new ArgumentException("Invalid external payload token.", nameof(token)); } + /// + /// Creates the ownership metadata stamped on every blob this store writes, so a later purge can prove the + /// store created the blob before deleting it. + /// + static Dictionary CreateOwnershipMetadata() => + new Dictionary(StringComparer.OrdinalIgnoreCase) + { + [OwnershipMarkerName] = OwnershipMarkerValue, + }; + + /// + /// Returns whether the supplied blob metadata carries this store's ownership marker. Azure treats metadata + /// names as case-insensitive, so the lookup is too. + /// + static bool HasOwnershipMarker(IDictionary? metadata) + { + if (metadata is null) + { + return false; + } + + foreach (KeyValuePair entry in metadata) + { + if (string.Equals(entry.Key, OwnershipMarkerName, StringComparison.OrdinalIgnoreCase)) + { + return string.Equals(entry.Value, OwnershipMarkerValue, StringComparison.Ordinal); + } + } + + return false; + } + static async Task WritePayloadAsync(byte[] payloadBuffer, Stream target, CancellationToken cancellationToken) { #if NETSTANDARD2_0 diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadDeleteOutcome.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadDeleteOutcome.cs new file mode 100644 index 00000000..532cbcd9 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadDeleteOutcome.cs @@ -0,0 +1,28 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask; + +/// +/// The outcome of deleting a payload through . All three values are +/// successful terminal outcomes; failures are surfaced as exceptions instead. +/// +public enum PayloadDeleteOutcome +{ + /// + /// The payload's backing object was deleted by this call. + /// + Deleted, + + /// + /// The payload's backing object was already absent. Deletion is idempotent, so this is a success. + /// + AlreadyAbsent, + + /// + /// The backing object exists but does not carry the store's ownership marker, so the store did not + /// create it and left it untouched. The payload reference is still resolved, because an object the + /// store never wrote is not the store's to delete. + /// + NotStoreOwned, +} diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs index b0fe6f80..ee6f22b2 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs @@ -24,6 +24,28 @@ 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. + /// Implementations must delete only objects they created; an object that carries no proof of the + /// store's ownership must be left untouched and reported as + /// . + /// + /// The opaque reference token. + /// Cancellation token. + /// + /// The outcome of the deletion: whether the object was deleted, was already absent, or was left in + /// place because the store does not own it. + /// + 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..080b697d 100644 --- a/src/Grpc/orchestrator_service.proto +++ b/src/Grpc/orchestrator_service.proto @@ -786,6 +786,19 @@ service TaskHubSidecarService { rpc PurgeInstances(PurgeInstancesRequest) returns (PurgeInstancesResponse); rpc GetWorkItems(GetWorkItemsRequest) returns (stream WorkItem); + + // Returns a bounded, deterministically ordered batch of due large-payload tombstones whose + // external blobs the worker must delete. Scoped to the caller's authenticated task hub. + // Only rows that are pending and whose next attempt time has arrived are returned; a row stays + // pending until its outcome is reported, so this is safe under retries and duplicate callers. + rpc GetLargePayloadTombstones(GetLargePayloadTombstonesRequest) returns (GetLargePayloadTombstonesResponse); + + // Reports the outcome of each attempted blob deletion. The backend owns retry scheduling and + // branches solely on `disposition`: it deletes rows reported as DELETED, reschedules RETRY on + // its own backoff, and moves QUARANTINED rows out of the active fetch while preserving their + // evidence. The worker never computes a retry delay. + rpc ReportLargePayloadPurgeResults(ReportLargePayloadPurgeResultsRequest) returns (ReportLargePayloadPurgeResultsResponse); + rpc CompleteActivityTask(ActivityResponse) returns (CompleteTaskResponse); rpc CompleteOrchestratorTask(OrchestratorResponse) returns (CompleteTaskResponse); rpc CompleteEntityTask(EntityBatchResult) returns (CompleteTaskResponse); @@ -825,6 +838,85 @@ service TaskHubSidecarService { rpc SkipGracefulOrchestrationTerminations(SkipGracefulOrchestrationTerminationsRequest) returns (SkipGracefulOrchestrationTerminationsResponse); } +// server -> client: one tombstoned large-payload row whose external blob the worker must delete. +message LargePayloadTombstone { + int32 partitionId = 1; + int64 instanceKey = 2; + int64 payloadId = 3; + + // A self-describing SDK v2 token: "blob:v2:{fullBlobUrl}". + // Legacy v1 tokens are never tombstoned: v1 carries a container name but not the storage + // account, so a delete against the configured account cannot be verified. The backend + // hard-deletes v1 payload rows instead. + string token = 4; + + // Optimistic-concurrency guard. The worker echoes this value back unmodified in the result so + // the backend can reject duplicate or stale reports without taking a per-row lease. + int64 revision = 5; +} + +// The outcome of a single blob deletion attempt. The split is by whether a failure can self-heal. +enum LargePayloadPurgeDisposition { + // Required: proto3 reserves 0 as the first value, and scalars have no field presence, so an + // unset field arrives as 0. Keeping 0 meaningless is load-bearing here: if 0 meant DELETED, a + // client that failed to set this field would make the backend delete tombstones and orphan the + // blobs permanently. The backend must reject a result carrying this value. + LARGE_PAYLOAD_PURGE_DISPOSITION_UNSPECIFIED = 0; + + // Terminal success: the tombstone is resolved and the backend deletes it. Covers the blob being + // deleted, the blob already being absent, and the blob being deliberately left in place because + // the payload store does not own it. All three are terminal because none of them can be + // improved by trying again. + LARGE_PAYLOAD_PURGE_DISPOSITION_DELETED = 1; + + // The failure may self-heal, so the row stays pending and the backend sets the next attempt. + LARGE_PAYLOAD_PURGE_DISPOSITION_RETRY = 2; + + // A deterministic failure or protocol violation that retrying can never fix. The row leaves the + // polling set but is never deleted or expired: it keeps the token, which after the payload row is + // gone is the only durable record of the blob, so discarding it would orphan the blob silently. + // Resolving a quarantined row is a deliberate operator action. Why it failed is not recorded here + // and is not meant to be; that detail lives in the worker's telemetry at full fidelity. + LARGE_PAYLOAD_PURGE_DISPOSITION_QUARANTINED = 3; +} + +// client -> server: the outcome of exactly one tombstoned row. +message LargePayloadPurgeResult { + // Row identity, echoed from the corresponding LargePayloadTombstone. + int32 partitionId = 1; + int64 instanceKey = 2; + int64 payloadId = 3; + + // Echoed unmodified from the fetched tombstone; used as a compare-and-swap guard. + int64 revision = 4; + + // The only field the backend acts on. Deliberately the only outcome field on this message: + // anything finer would be write-only. Failure detail stays in the worker's own telemetry, which + // holds the full exception rather than a lossy classification, and a row is correlated to it by + // (partitionId, instanceKey, payloadId). + LargePayloadPurgeDisposition disposition = 5; +} + +// client -> server: request up to `limit` due tombstones for the caller's task hub. +message GetLargePayloadTombstonesRequest { + // The maximum number of rows to return. The service clamps this to its own maximum. + int32 limit = 1; +} + +// server -> client: the due tombstones whose blobs the worker must delete. +message GetLargePayloadTombstonesResponse { + repeated LargePayloadTombstone tombstones = 1; +} + +// client -> server: a bounded batch of purge outcomes. +message ReportLargePayloadPurgeResultsRequest { + repeated LargePayloadPurgeResult results = 1; +} + +// server -> client: acknowledgement that the reported outcomes were recorded. +message ReportLargePayloadPurgeResultsResponse { +} + message GetWorkItemsRequest { int32 maxConcurrentOrchestrationWorkItems = 1; int32 maxConcurrentActivityWorkItems = 2; @@ -832,6 +924,16 @@ message GetWorkItemsRequest { repeated WorkerCapability capabilities = 10; WorkItemFilters workItemFilters = 11; + + // Task-hub scoped opt-in for large-payload blob auto-purge. + // + // This is a setting, not a WorkerCapability: WORKER_CAPABILITY_LARGE_PAYLOADS means a worker + // *can resolve* externalized payloads, not that the customer opted into *deleting* them. + // A capability list is presence-only and cannot express an explicit false, so a three-state + // wrapper is used instead: + // absent -> the worker has no opinion (older SDK); the persisted setting is unchanged. + // true/false -> an explicit customer choice; persisted only when the value actually differs. + google.protobuf.BoolValue large_payload_auto_purge_enabled = 12; } enum WorkerCapability { diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 5dd18d52..ea5f7c87 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -349,6 +349,10 @@ async ValueTask BuildRuntimeStateAsync( workerOptions.Concurrency.MaximumConcurrentEntityWorkItems, Capabilities = { this.worker.grpcOptions.Capabilities }, WorkItemFilters = this.worker.workItemFilters?.ToGrpcWorkItemFilters(), + + // Left unset when the worker has no opinion, so the backend can distinguish "not + // configured" from an explicit opt-out. + LargePayloadAutoPurgeEnabled = this.worker.grpcOptions.LargePayloadAutoPurgeEnabled, }, cancellationToken: cancellation); } diff --git a/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs b/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs index 59c21a00..a7f5fa34 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs @@ -44,6 +44,13 @@ public sealed class GrpcDurableTaskWorkerOptions : DurableTaskWorkerOptions /// public HashSet Capabilities { get; } = new() { P.WorkerCapability.HistoryStreaming }; + /// + /// Gets or sets a value indicating whether this worker opts in to large-payload auto-purge, announced to + /// the backend on connection. null means the worker expresses no opinion and the backend keeps its + /// current behavior; an explicit true or false is the customer's choice. + /// + public bool? LargePayloadAutoPurgeEnabled { get; set; } + /// /// Gets or sets the maximum size of all actions in a complete orchestration work item chunk. /// The default value is 3.9MB. We leave some headroom to account for request size overhead. diff --git a/test/Client/Grpc.Tests/LargePayloadPurgeEnumParityTests.cs b/test/Client/Grpc.Tests/LargePayloadPurgeEnumParityTests.cs new file mode 100644 index 00000000..48658410 --- /dev/null +++ b/test/Client/Grpc.Tests/LargePayloadPurgeEnumParityTests.cs @@ -0,0 +1,65 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Reflection; +using Microsoft.DurableTask.Client; +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Client.Grpc.Tests; + +/// +/// maps the managed purge disposition +/// onto its protobuf counterpart by numeric value rather than by name, which is only correct while the two +/// sides agree on every value. A silent drift would not fail to compile; it would send the backend a different +/// disposition than the worker decided and delete or quarantine the wrong rows. These tests pin the mapping. +/// +public class LargePayloadPurgeEnumParityTests +{ + [Fact] + public void Disposition_ManagedAndProtobufValues_AreIdentical() + { + // Arrange & Act + Dictionary managed = Enum.GetValues(typeof(LargePayloadPurgeDisposition)) + .Cast() + .ToDictionary(v => (int)v, v => v.ToString()); + Dictionary proto = Enum.GetValues(typeof(P.LargePayloadPurgeDisposition)) + .Cast() + .ToDictionary(v => (int)v, v => v.ToString()); + + // Assert - same numeric values AND the same names at each value, so neither side can gain, lose, or + // renumber a member unnoticed. + managed.Should().Equal(proto); + } + + /// + /// The numeric cast is safe only because no enum crosses the wire inbound on this feature: the SDK + /// casts a value it defined itself, so it can never receive an unknown value and silently reinterpret it. + /// That invariant holds today by the shape of the contract, not by construction, and nothing in the code + /// states it. Adding an enum to an inbound type would create exactly that path - a newer backend sending a + /// value this SDK does not know, mapped by raw numeric cast onto a valid-but-wrong member - and it would + /// compile silently. This pins the invariant so it fails here instead. + /// + /// A type carrying server-to-client data for the purge feature. + [Theory] + [InlineData(typeof(LargePayloadTombstone))] + [InlineData(typeof(P.LargePayloadTombstone))] + [InlineData(typeof(P.GetLargePayloadTombstonesResponse))] + [InlineData(typeof(P.ReportLargePayloadPurgeResultsResponse))] + public void InboundTypes_ExposeNoEnumMembers(Type inboundType) + { + // Arrange & Act + List enumMembers = inboundType + .GetProperties(BindingFlags.Public | BindingFlags.Instance) + .Where(p => (Nullable.GetUnderlyingType(p.PropertyType) ?? p.PropertyType).IsEnum) + .Select(p => $"{inboundType.Name}.{p.Name}") + .ToList(); + + // Assert + enumMembers.Should().BeEmpty( + "an enum on an inbound type invalidates the numeric enum cast in " + + "GrpcDurableTaskClient.ReportLargePayloadPurgeResultsAsync. The SDK would map a value chosen by the " + + "backend - including one a newer backend added that this SDK does not know - onto a managed member " + + "by raw numeric value, silently mis-dispositioning rows. Map inbound enums explicitly instead, with " + + "a switch that handles unknown values"); + } +} diff --git a/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobOrchestratorTests.cs b/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobOrchestratorTests.cs new file mode 100644 index 00000000..609cdcb9 --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobOrchestratorTests.cs @@ -0,0 +1,113 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using FluentAssertions; +using Microsoft.DurableTask.AzureBlobPayloads; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Entities; +using Xunit; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests.AutoPurge; + +public class BlobPurgeJobOrchestratorTests +{ + static readonly EntityInstanceId JobEntityId = new(nameof(BlobPurgeJob), BlobPurgeConstants.JobId); + + readonly List requested = []; + readonly TestLogger logger = new(); + + [Fact] + public async Task RunAsync_UsesBatchSizeFromEntity_NotFromInput() + { + // Arrange - the orchestrator is perpetual, so its input is fixed at creation and carried verbatim + // through every continue-as-new. Reading the batch size from the input would pin the value written by + // the very first Create for the life of the job, which is what made a configuration change impossible + // to apply. The entity is the authority. + Mock context = this.ContextFor( + new BlobPurgeJobState { Status = BlobPurgeJobStatus.Active, PurgeBatchSize = 777 }); + + // Act + await new BlobPurgeJobOrchestrator().RunAsync( + context.Object, new BlobPurgeJobRunRequest(JobEntityId, PurgeBatchSize: 100)); + + // Assert + this.AssertNoCycleFailed(); + this.requested.Should().Equal(777); + } + + [Fact] + public async Task RunAsync_WhenEntityBatchSizeIsUnset_FallsBackToInput() + { + // Arrange - an entity written by an older build carries no batch size at all. Passing that zero to the + // fetch activity would ask the backend for nothing on every cycle: a silent, permanent stall. + Mock context = this.ContextFor( + new BlobPurgeJobState { Status = BlobPurgeJobStatus.Active, PurgeBatchSize = 0 }); + + // Act + await new BlobPurgeJobOrchestrator().RunAsync( + context.Object, new BlobPurgeJobRunRequest(JobEntityId, PurgeBatchSize: 100)); + + // Assert + this.AssertNoCycleFailed(); + this.requested.Should().Equal(100); + } + + [Fact] + public async Task RunAsync_WhenJobIsStopped_ExitsWithoutFetching() + { + // Arrange - Stop moves the entity off Active without touching this orchestrator, so the exit is the + // orchestrator's own decision on its own schedule. Nothing must be fetched on a stopped job. + Mock context = this.ContextFor( + new BlobPurgeJobState { Status = BlobPurgeJobStatus.Pending, PurgeBatchSize = 777 }); + + // Act + await new BlobPurgeJobOrchestrator().RunAsync( + context.Object, new BlobPurgeJobRunRequest(JobEntityId, PurgeBatchSize: 100)); + + // Assert - an empty fetch list is also what a broken mock produces, so the stopping log is asserted + // too: it is the only evidence that the orchestrator read the state and chose to exit. + this.AssertNoCycleFailed(); + this.requested.Should().BeEmpty(); + this.logger.Logs.Should().Contain(entry => entry.Message.Contains("stopping")); + } + + /// + /// Guards against the whole test passing through the orchestrator's catch-all cycle handler, which would + /// leave every observation empty and make the assertions vacuous. + /// + void AssertNoCycleFailed() => + this.logger.Logs.Should().NotContain(entry => entry.Message.Contains("cycle for job")); + + /// + /// Builds a context whose first entity read returns and whose second returns a + /// stopped job, so the perpetual loop runs at most one cycle and then exits. Batch sizes passed to the + /// fetch activity are recorded. + /// + Mock ContextFor(BlobPurgeJobState first) + { + Mock context = new(); + Mock entities = new(); + + context.Setup(c => c.Entities).Returns(entities.Object); + context.Setup(c => c.CreateReplaySafeLogger()).Returns(this.logger); + context.Setup(c => c.CreateTimer(It.IsAny(), It.IsAny())) + .Returns(Task.CompletedTask); + + entities + .SetupSequence(e => e.CallEntityAsync( + It.IsAny(), + nameof(BlobPurgeJob.Get), + It.IsAny(), + It.IsAny())) + .ReturnsAsync(first) + .ReturnsAsync(new BlobPurgeJobState { Status = BlobPurgeJobStatus.Pending }); + + context + .Setup(c => c.CallActivityAsync>( + It.IsAny(), It.IsAny(), It.IsAny())) + .Callback((_, input, _) => this.requested.Add((int)input!)) + .ReturnsAsync([]); + + return context; + } +} diff --git a/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobStarterTests.cs b/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobStarterTests.cs new file mode 100644 index 00000000..83d51fb2 --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobStarterTests.cs @@ -0,0 +1,395 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using FluentAssertions; +using Microsoft.DurableTask.AzureBlobPayloads; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.Entities; +using Microsoft.DurableTask.Entities; +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_WhenAutoPurgeDisabledAndJobIsActive_SignalsJobToStop() + { + // Arrange - auto-purge is off while a job is running. Returning silently was the whole disable bug: the + // job is a perpetual orchestrator owned by the task hub, so one created while the flag was on keeps + // deleting blobs no matter how many hosts start with it off. The disable path has to reach the backend. + BlobPayloadStore store = new(new LargePayloadStorageOptions("UseDevelopmentStorage=true")); + + // Act + (( EntityInstanceId Id, string Operation)? Signal, int Reads) run = + await RunDisabledStarterAsync(store, () => MetadataFor(BlobPurgeJobStatus.Active)); + + // Assert + run.Signal.Should().NotBeNull("the disable path must tell a running job to stop"); + run.Signal!.Value.Id.Should().Be(new EntityInstanceId(nameof(BlobPurgeJob), BlobPurgeConstants.JobId)); + run.Signal.Value.Operation.Should().Be(nameof(BlobPurgeJob.Stop)); + } + + [Fact] + public async Task StartAsync_WhenAutoPurgeDisabledAndNoJobExists_DoesNotSignal() + { + // Arrange - the common case by far: an app that externalizes payloads and never enabled auto-purge. + // Signalling regardless would create the job entity, because the framework persists entity state after + // every operation, so an app that never used the feature would still carry one entity per task hub. + BlobPayloadStore store = new(new LargePayloadStorageOptions("UseDevelopmentStorage=true")); + + // Act + (( EntityInstanceId Id, string Operation)? Signal, int Reads) run = + await RunDisabledStarterAsync(store, () => null); + + // Assert - Reads is the positive control. A null signal is also what a mock that was never reached + // produces, so the assertion that nothing was signalled is only meaningful alongside proof that the + // code did run and did ask. The same wiring signals when the job is active, which the test above pins. + run.Reads.Should().Be(1, "the starter must consult the job state rather than skipping the path entirely"); + run.Signal.Should().BeNull("there is no job to stop, so nothing may be created"); + } + + [Fact] + public async Task StartAsync_WhenAutoPurgeDisabledAndJobIsAlreadyStopped_DoesNotSignal() + { + // Arrange - a job that has already been stopped needs nothing. The entity would ignore the signal, but + // the framework still persists state on every operation, so signalling would write on every host start. + BlobPayloadStore store = new(new LargePayloadStorageOptions("UseDevelopmentStorage=true")); + + // Act + (( EntityInstanceId Id, string Operation)? Signal, int Reads) run = + await RunDisabledStarterAsync(store, () => MetadataFor(BlobPurgeJobStatus.Pending)); + + // Assert + run.Reads.Should().Be(1); + run.Signal.Should().BeNull("a job that is not active must not be signalled on every restart"); + } + + [Fact] + public async Task StartAsync_WhenAutoPurgeDisabledAndEntityHasNoState_DoesNotSignal() + { + // Arrange - an entity whose state has been cleared still reports as existing until entity storage is + // cleaned, and its metadata carries no state at all. Reading EntityMetadata.State in that condition + // throws, so the status has to be reached through IncludesState. Such an entity has no running job. + BlobPayloadStore store = new(new LargePayloadStorageOptions("UseDevelopmentStorage=true")); + + // Act + (( EntityInstanceId Id, string Operation)? Signal, int Reads) run = await RunDisabledStarterAsync( + store, + () => new EntityMetadata( + new EntityInstanceId(nameof(BlobPurgeJob), BlobPurgeConstants.JobId))); + + // Assert + run.Reads.Should().Be(1); + run.Signal.Should().BeNull("an entity with no state has no running job"); + } + + [Fact] + public async Task StartAsync_WhenAutoPurgeDisabledAndStateCannotBeRead_SignalsJobToStopAnyway() + { + // Arrange - the pre-check is an optimization; stopping the job is the correctness-critical outcome. A + // client that cannot answer the query must not be able to prevent the stop, or the optimization's own + // failure mode would defeat the fix it is attached to. Retrying instead of falling through would be + // just as bad for a permanent failure: the loop would spin and the running job would never be told. + BlobPayloadStore store = new(new LargePayloadStorageOptions("UseDevelopmentStorage=true")); + + // Act + (( EntityInstanceId Id, string Operation)? Signal, int Reads) run = await RunDisabledStarterAsync( + store, + () => throw new NotSupportedException("entity queries are not supported")); + + // Assert + run.Signal.Should().NotBeNull("an unreadable state must not suppress the stop"); + run.Signal!.Value.Operation.Should().Be(nameof(BlobPurgeJob.Stop)); + } + + [Fact] + public async Task StartAsync_WhenAutoPurgeDisabledAndStoreCannotDelete_StillSignalsJobToStop() + { + // Arrange - the store-capability gate deliberately does not apply to the disable path. Stopping a job + // requires no ability to delete anything, and a user who has switched to a store that cannot delete is + // precisely the user whose still-running job must be stopped. + + // Act + (( EntityInstanceId Id, string Operation)? Signal, int Reads) run = await RunDisabledStarterAsync( + new NonDeletingPayloadStore(), () => MetadataFor(BlobPurgeJobStatus.Active)); + + // Assert + run.Signal.Should().NotBeNull("a store that cannot delete must not block stopping the job"); + run.Signal!.Value.Operation.Should().Be(nameof(BlobPurgeJob.Stop)); + } + + [Fact] + public async Task EnsureJob_SchedulesBridge_DedupingOnlyPendingAndRunning() + { + // Arrange - the dedupe list is an inverted whitelist: the wire policy is (all statuses - dedupe), so a + // status omitted from the call silently becomes replaceable. Nothing in the compiler or the type system + // catches that, so the exact set is pinned here. + BlobPayloadStore store = new(new LargePayloadStorageOptions("UseDevelopmentStorage=true")); + Mock client = new("test"); + TaskCompletionSource scheduled = new(); + client + .Setup(c => c.ScheduleNewOrchestrationInstanceAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback( + (_, _, options, _) => scheduled.TrySetResult(options)) + .ReturnsAsync(BlobPurgeConstants.StarterInstanceId); + + Mock provider = new(); + provider.Setup(p => p.GetClient(It.IsAny())).Returns(client.Object); + BlobPurgeJobStarter starter = new( + provider.Object, + store, + OptionsFor(new LargePayloadStorageOptions("UseDevelopmentStorage=true") { AutoPurge = true }), + "test", + new TestLogger()); + + // Act - the ensure work runs on a background task, so wait for the scheduling call rather than assuming + // it already happened. + await starter.StartAsync(CancellationToken.None); + Task completed = await Task.WhenAny(scheduled.Task, Task.Delay(TimeSpan.FromSeconds(30))); + await starter.StopAsync(CancellationToken.None); + + // Assert + completed.Should().BeSameAs(scheduled.Task, "the starter must schedule the bridge orchestration"); + StartOrchestrationOptions? options = await scheduled.Task; + options.Should().NotBeNull(); + options!.InstanceId.Should().Be(BlobPurgeConstants.StarterInstanceId); + + // Every status other than Pending and Running is replaceable, so a finished bridge is re-run on the + // next reconciliation pass. That is what lets the job rebuild itself after the entity is removed. The + // set is asserted exactly, never as a superset, because the hazard is a silent omission. + options.DedupeStatuses.Should().BeEquivalentTo(["Pending", "Running"]); + } + + [Fact] + public async Task EnsureJob_AfterSuccessfulSchedule_KeepsReconciling() + { + // Arrange - the loop used to return as soon as one schedule succeeded, which left recovery bounded by + // host starts: this call is the only thing that re-signals the job's Run, so an orchestrator that died + // mid-lifetime stayed down until the process was restarted. It has to keep re-issuing the call instead. + Mock client = new("test"); + TaskCompletionSource secondSchedule = new(); + int schedules = 0; + client + .Setup(c => c.ScheduleNewOrchestrationInstanceAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback(() => + { + if (Interlocked.Increment(ref schedules) >= 2) + { + secondSchedule.TrySetResult(true); + } + }) + .ReturnsAsync(BlobPurgeConstants.StarterInstanceId); + + // A short interval rather than a fake clock. The test still gates on the second call happening, never + // on time having passed, so it neither sleeps nor assumes anything about how long a pass takes. + BlobPurgeJobStarter starter = EnabledStarterFor(client.Object); + starter.ReconcileInterval = TimeSpan.FromMilliseconds(10); + + // Act + await starter.StartAsync(CancellationToken.None); + Task completed = await Task.WhenAny(secondSchedule.Task, Task.Delay(TimeSpan.FromSeconds(30))); + await starter.StopAsync(CancellationToken.None); + + // Assert - a loop that returned after its first success leaves this waiting until the timeout fires. + completed.Should().BeSameAs( + secondSchedule.Task, "reconciliation must repeat rather than ending after the first success"); + } + + [Fact] + public async Task StopAsync_WhileWaitingForNextReconcile_ReturnsPromptly() + { + // Arrange - the default interval, so the loop parks in a five-minute wait once its first pass succeeds. + // StopAsync awaits the background task, so a wait that did not observe cancellation would hold host + // shutdown for the rest of the interval. That is the one real regression risk in making the loop + // periodic, and it is invisible in the other tests because they would simply hang. + Mock client = new("test"); + TaskCompletionSource scheduled = new(); + client + .Setup(c => c.ScheduleNewOrchestrationInstanceAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback(() => scheduled.TrySetResult(true)) + .ReturnsAsync(BlobPurgeConstants.StarterInstanceId); + + BlobPurgeJobStarter starter = EnabledStarterFor(client.Object); + + // Act - waiting for the first schedule is the positive control. Without it a prompt StopAsync would + // also be what a starter that never reached the loop at all produces. + await starter.StartAsync(CancellationToken.None); + Task firstPass = await Task.WhenAny(scheduled.Task, Task.Delay(TimeSpan.FromSeconds(30))); + Task stop = starter.StopAsync(CancellationToken.None); + Task stopped = await Task.WhenAny(stop, Task.Delay(TimeSpan.FromSeconds(30))); + + // Assert + firstPass.Should().BeSameAs(scheduled.Task, "the loop must have reached the wait being cancelled here"); + stopped.Should().BeSameAs(stop, "shutdown must cancel the wait rather than sit out the interval"); + } + + static IOptionsMonitor OptionsFor(LargePayloadStorageOptions options) + { + Mock> monitor = new(); + monitor.Setup(m => m.Get(It.IsAny())).Returns(options); + return monitor.Object; + } + + /// + /// Builds a starter with auto-purge enabled over the given client. The real blob store is used because the + /// store-capability gate refuses to start the job for any other kind, so a stub would never reach the + /// ensure loop. UseDevelopmentStorage=true constructs it offline, with no network I/O. + /// + static BlobPurgeJobStarter EnabledStarterFor(DurableTaskClient client) + { + Mock provider = new(); + provider.Setup(p => p.GetClient(It.IsAny())).Returns(client); + + return new BlobPurgeJobStarter( + provider.Object, + new BlobPayloadStore(new LargePayloadStorageOptions("UseDevelopmentStorage=true")), + OptionsFor(new LargePayloadStorageOptions("UseDevelopmentStorage=true") { AutoPurge = true }), + "test", + new TestLogger()); + } + + /// + /// Builds entity metadata carrying a job in the given status. + /// + static EntityMetadata MetadataFor(BlobPurgeJobStatus status) => + new( + new EntityInstanceId(nameof(BlobPurgeJob), BlobPurgeConstants.JobId), + new BlobPurgeJobState { Status = status }); + + /// + /// Runs a starter with auto-purge disabled, using as the job state the client + /// reports (it may return null for an absent entity, or throw to simulate a client that cannot answer). + /// Returns the entity signal the starter emitted, if any, and how many times the state was read. + /// + static async Task<(( EntityInstanceId Id, string Operation)? Signal, int Reads)> RunDisabledStarterAsync( + PayloadStore store, Func?> read) + { + Mock entities = new("test"); + TaskCompletionSource readCalled = new(); + (EntityInstanceId Id, string Operation)? signal = null; + int reads = 0; + + // The three-argument overload is set up because that is the one the starter calls. The shorter + // GetEntityAsync(id, cancellation) is virtual, so a mock overrides it instead of forwarding, and a + // setup placed on the abstract method alone would silently never match. + entities + .Setup(e => e.GetEntityAsync( + It.IsAny(), It.IsAny(), It.IsAny())) + .Returns(() => + { + reads++; + readCalled.TrySetResult(true); + return Task.FromResult(read()); + }); + + entities + .Setup(e => e.SignalEntityAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback( + (id, operation, _, _, _) => signal = (id, operation)) + .Returns(Task.CompletedTask); + + Mock client = new("test"); + client.Setup(c => c.Entities).Returns(entities.Object); + Mock provider = new(); + provider.Setup(p => p.GetClient(It.IsAny())).Returns(client.Object); + + BlobPurgeJobStarter starter = new( + provider.Object, + store, + OptionsFor(new LargePayloadStorageOptions("UseDevelopmentStorage=true") { AutoPurge = false }), + "test", + new TestLogger()); + + // The work runs on a background task, so wait for the read before shutting down. StopAsync then waits + // for that task to finish, which makes the observation deterministic: by the time it returns the + // starter has either signalled or decided not to, rather than being timed out mid-decision. + await starter.StartAsync(CancellationToken.None); + await Task.WhenAny(readCalled.Task, Task.Delay(TimeSpan.FromSeconds(30))); + await starter.StopAsync(CancellationToken.None); + + return (signal, reads); + } + + 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..88d4fa6a --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs @@ -0,0 +1,370 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using FluentAssertions; +using Microsoft.DurableTask.AzureBlobPayloads; +using Microsoft.DurableTask.Entities; +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(); + + // Starting the job means signalling Run, and this path has always done so. Asserted explicitly because + // the already-active path now signals Run too, which makes this the case that would silently stop being + // covered if the two branches were ever collapsed. + Mock.Get(operation.Context).Verify( + c => c.SignalEntity( + It.IsAny(), + nameof(BlobPurgeJob.Run), + It.IsAny(), + It.IsAny()), + Times.Once); + } + + [Fact] + public async Task Create_WhenAlreadyActive_UpdatesBatchSizeAndReSignalsRun() + { + // Arrange - the job is already running and the configured batch size has changed. Create is the only + // path by which a new batch size can reach an active job, so it must be taken. + 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 + BlobPurgeJobState state = Assert.IsType( + operation.State.GetState(typeof(BlobPurgeJobState))); + state.Status.Should().Be(BlobPurgeJobStatus.Active); + state.PurgeBatchSize.Should().Be(999); + + // Run is re-signalled even though the job is already active. That is what lets a job whose orchestrator + // has died be rebuilt at the next reconciliation pass, and it is safe because the backend discards the + // resulting start while the orchestrator is alive rather than replacing it. + Mock.Get(operation.Context).Verify( + c => c.SignalEntity( + It.IsAny(), + nameof(BlobPurgeJob.Run), + It.IsAny(), + It.IsAny()), + Times.Once); + } + + [Fact] + public async Task Create_WhenAlreadyActive_SignalsNothingOtherThanRun() + { + // Arrange - pins that re-signalling Run is the only signal the already-active path emits. Verifying the + // Run signal alone would still pass if a second, different signal were added beside it. + 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 - the Times.Once above is the positive control for this Times.Never: both target the same + // four-argument overload on the same mock, so this cannot be passing because nothing was recorded. + Mock.Get(operation.Context).Verify( + c => c.SignalEntity( + It.IsAny(), + It.Is(name => name != nameof(BlobPurgeJob.Run)), + It.IsAny(), + It.IsAny()), + Times.Never); + } + + [Fact] + public async Task Create_WhenAlreadyActive_AndBatchSizeUnchanged_DoesNotMoveLastModifiedAt() + { + // Arrange - the steady state. Create runs on every reconciliation pass, and almost every one of those + // carries the same configured batch size the job already has. If that rewrote LastModifiedAt, the field + // would degrade to "time of the last pass" and say nothing about the job. + DateTimeOffset configuredAt = DateTimeOffset.UtcNow.AddDays(-2); + BlobPurgeJobState existing = new() + { + Status = BlobPurgeJobStatus.Active, + LastModifiedAt = configuredAt, + PurgeBatchSize = 250, + }; + TestEntityOperation operation = new( + nameof(BlobPurgeJob.Create), + new TestEntityState(existing), + 250); + + // Act + await this.job.RunAsync(operation); + + // Assert - the sibling test below is the positive control: identical wiring, differing only in the + // batch size passed in, and it proves this same path does move the field when something changes. + BlobPurgeJobState state = Assert.IsType( + operation.State.GetState(typeof(BlobPurgeJobState))); + state.LastModifiedAt.Should().Be(configuredAt); + state.PurgeBatchSize.Should().Be(250); + } + + [Fact] + public async Task Create_WhenAlreadyActive_AndBatchSizeChanged_MovesLastModifiedAt() + { + // Arrange - a real configuration change reaching an active job, which is the one thing this path + // exists to deliver and the one case that must be recorded. + DateTimeOffset configuredAt = DateTimeOffset.UtcNow.AddDays(-2); + BlobPurgeJobState existing = new() + { + Status = BlobPurgeJobStatus.Active, + LastModifiedAt = configuredAt, + PurgeBatchSize = 250, + }; + TestEntityOperation operation = new( + nameof(BlobPurgeJob.Create), + new TestEntityState(existing), + 500); + + // Act + await this.job.RunAsync(operation); + + // Assert + BlobPurgeJobState state = Assert.IsType( + operation.State.GetState(typeof(BlobPurgeJobState))); + state.PurgeBatchSize.Should().Be(500); + state.LastModifiedAt.Should().BeAfter(configuredAt); + } + + [Fact] + public async Task Run_DoesNotMoveLastModifiedAt() + { + // Arrange - Run schedules an orchestrator and changes nothing about the job. It is signalled by every + // Create, so writing here would move the field on every reconciliation pass and undo the conditional + // write above. + DateTimeOffset configuredAt = DateTimeOffset.UtcNow.AddDays(-2); + BlobPurgeJobState existing = new() + { + Status = BlobPurgeJobStatus.Active, + LastModifiedAt = configuredAt, + PurgeBatchSize = 250, + }; + TestEntityOperation operation = new( + nameof(BlobPurgeJob.Run), + new TestEntityState(existing), + null); + + // Act + await this.job.RunAsync(operation); + + // Assert - scheduling is asserted first as the positive control. Without it a misdispatched operation + // would write nothing and pass this test for entirely the wrong reason. + Mock.Get(operation.Context).Verify( + c => c.ScheduleNewOrchestration( + It.IsAny(), + It.IsAny(), + It.IsAny()), + Times.Once); + + BlobPurgeJobState state = Assert.IsType( + operation.State.GetState(typeof(BlobPurgeJobState))); + state.LastModifiedAt.Should().Be(configuredAt); + } + + [Fact] + public async Task Run_WhenActive_SchedulesOrchestratorAtTheFixedInstanceId() + { + // Arrange - the fixed instance ID is the mechanism the whole restart story rests on. It is what lets the + // backend recognize a start as targeting the existing orchestrator, and therefore discard it while that + // orchestrator is alive instead of running a second one alongside it. + BlobPurgeJobState existing = new() + { + Status = BlobPurgeJobStatus.Active, + PurgeBatchSize = 250, + }; + TestEntityOperation operation = new( + nameof(BlobPurgeJob.Run), + new TestEntityState(existing), + null); + Mock.Get(operation.Context) + .Setup(c => c.Id) + .Returns(new EntityInstanceId(nameof(BlobPurgeJob), BlobPurgeConstants.JobId)); + + // Act + await this.job.RunAsync(operation); + + // Assert + Mock.Get(operation.Context).Verify( + c => c.ScheduleNewOrchestration( + It.IsAny(), + It.IsAny(), + It.Is(o => + o.InstanceId == BlobPurgeConstants.GetOrchestratorInstanceId(BlobPurgeConstants.JobId))), + Times.Once); + } + + [Fact] + public async Task Run_WhenNotActive_SchedulesNothing() + { + // Arrange - a Run signal arriving after the job was stopped. Create signals Run rather than starting the + // orchestrator itself, so a Stop landing between the two leaves this signal in flight against a job that + // must no longer purge. This guard is what makes the stop win instead of the stale signal restarting it. + BlobPurgeJobState existing = new() + { + Status = BlobPurgeJobStatus.Pending, + PurgeBatchSize = 250, + }; + TestEntityOperation operation = new( + nameof(BlobPurgeJob.Run), + new TestEntityState(existing), + null); + + // Act + await this.job.RunAsync(operation); + + // Assert - the Times.Once above is the positive control: same mocked type, same three-argument overload, + // so this cannot be passing merely because the mock records nothing. + Mock.Get(operation.Context).Verify( + c => c.ScheduleNewOrchestration( + It.IsAny(), + It.IsAny(), + It.IsAny()), + Times.Never); + } + + [Fact] + public async Task Stop_WhenActive_MovesToPendingAndKeepsHistory() + { + // Arrange - a running job. Stopping must not discard the configuration or the progress counters: they + // are wanted if the job is started again, and CreatedAt is what distinguishes a stopped job from one + // that was never started. + DateTimeOffset createdAt = DateTimeOffset.UtcNow.AddDays(-3); + BlobPurgeJobState existing = new() + { + Status = BlobPurgeJobStatus.Active, + CreatedAt = createdAt, + PurgedCount = 17, + PurgeBatchSize = 250, + }; + TestEntityOperation operation = new( + nameof(BlobPurgeJob.Stop), + new TestEntityState(existing), + null); + + // Act + await this.job.RunAsync(operation); + + // Assert + BlobPurgeJobState state = Assert.IsType( + operation.State.GetState(typeof(BlobPurgeJobState))); + state.Status.Should().Be(BlobPurgeJobStatus.Pending); + state.CreatedAt.Should().Be(createdAt); + state.PurgedCount.Should().Be(17); + state.PurgeBatchSize.Should().Be(250); + state.LastModifiedAt.Should().NotBeNull(); + } + + [Fact] + public async Task Stop_WhenNotActive_LeavesStateUntouched() + { + // Arrange - a job that is already stopped. The starter's client-side pre-check normally suppresses a + // redundant stop, so this is the signal that races past it: the job stopped between that read and this + // signal landing. Rewriting LastModifiedAt here would report the losing side of that race as if it + // were the moment the job stopped. + DateTimeOffset stoppedAt = DateTimeOffset.UtcNow.AddHours(-6); + BlobPurgeJobState existing = new() + { + Status = BlobPurgeJobStatus.Pending, + CreatedAt = DateTimeOffset.UtcNow.AddDays(-3), + LastModifiedAt = stoppedAt, + PurgeBatchSize = 250, + }; + TestEntityOperation operation = new( + nameof(BlobPurgeJob.Stop), + new TestEntityState(existing), + null); + + // Act + await this.job.RunAsync(operation); + + // Assert + BlobPurgeJobState state = Assert.IsType( + operation.State.GetState(typeof(BlobPurgeJobState))); + state.Status.Should().Be(BlobPurgeJobStatus.Pending); + state.LastModifiedAt.Should().Be(stoppedAt); + state.PurgeBatchSize.Should().Be(250); + } + + [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..94ad4830 --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/AutoPurge/DeleteExternalBlobActivityTests.cs @@ -0,0 +1,258 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Azure; +using FluentAssertions; +using Microsoft.DurableTask.AzureBlobPayloads; +using Microsoft.DurableTask.Client; +using Xunit; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests.AutoPurge; + +/// +/// The reported outcome carries the disposition alone, so each case pins two things: the disposition, which is +/// what the backend acts on, and the logged cause, which is now the only record of why an attempt reached that +/// disposition. Several branches share a disposition and are told apart only by their cause, so asserting the +/// disposition alone would let two branches collapse into one unnoticed. +/// +public class DeleteExternalBlobActivityTests +{ + const string V2Token = "blob:v2:https://acct.blob.core.windows.net/payloads/abc123"; + + [Fact] + public async Task RunAsync_WhenDeleteThrowsRequestFailed400_Quarantines() + { + // Arrange - a Status 400 (e.g. InvalidResourceName) is a permanent service rejection. + StubPayloadStore store = new(new RequestFailedException(400, "bad", "InvalidResourceName", null)); + TestLogger logger = new(); + DeleteExternalBlobActivity activity = new(store, logger); + + // Act + BlobPurgeOutcome outcome = await activity.RunAsync(null!, V2Token); + + // Assert - quarantined (evidence preserved), never a success-shaped discard. The sanitized storage + // error code is logged alongside the cause and is what distinguishes this from the parse failures. + outcome.Disposition.Should().Be(LargePayloadPurgeDisposition.Quarantined); + logger.Logs.Should().ContainSingle( + l => l.Message.Contains("InvalidStorageRequest") && l.Message.Contains("InvalidResourceName")); + } + + [Fact] + public async Task RunAsync_WhenDeleteThrowsRequestFailedNon400_RetriesAsTransient() + { + // Arrange - a Status 503 that escaped the SDK's internal retries is still treated as transient. + StubPayloadStore store = new(new RequestFailedException(503, "busy", "ServerBusy", null)); + TestLogger logger = new(); + DeleteExternalBlobActivity activity = new(store, logger); + + // Act + BlobPurgeOutcome outcome = await activity.RunAsync(null!, V2Token); + + // Assert + outcome.Disposition.Should().Be(LargePayloadPurgeDisposition.Retry); + logger.Logs.Should().ContainSingle( + l => l.Message.Contains("TransientStorageFailure") && l.Message.Contains("ServerBusy")); + } + + [Theory] + [InlineData(401)] + [InlineData(403)] + public async Task RunAsync_WhenDeleteThrowsAuthorizationFailure_Retries(int status) + { + // Arrange - authorization can be fixed by reconfiguration, so it stays recoverable. + StubPayloadStore store = new(new RequestFailedException(status, "denied", "AuthorizationFailure", null)); + TestLogger logger = new(); + DeleteExternalBlobActivity activity = new(store, logger); + + // Act + BlobPurgeOutcome outcome = await activity.RunAsync(null!, V2Token); + + // Assert + outcome.Disposition.Should().Be(LargePayloadPurgeDisposition.Retry); + logger.Logs.Should().ContainSingle(l => l.Message.Contains("StorageAuthorizationFailed")); + } + + [Fact] + public async Task RunAsync_WhenDeleteThrowsPayloadStorageException_Retries() + { + // Arrange - the payload lives in a storage account the configured credential cannot reach. That is + // recoverable after a configuration or credential change, so it is deferred rather than discarded. + StubPayloadStore store = new(new PayloadStorageException("cross-account delete requires identity auth")); + TestLogger logger = new(); + DeleteExternalBlobActivity activity = new(store, logger); + + // Act + BlobPurgeOutcome outcome = await activity.RunAsync( + null!, "blob:v2:https://other.blob.core.windows.net/c/abc123"); + + // Assert - the cause is what separates this from the other retryable storage failures, which the + // contract no longer distinguishes. + outcome.Disposition.Should().Be(LargePayloadPurgeDisposition.Retry); + logger.Logs.Should().ContainSingle(l => l.Message.Contains("StorageAccountUnreachable")); + } + + [Fact] + public async Task RunAsync_V1Token_QuarantinesWithoutCallingStore() + { + // Arrange - a v1 token names a container but not the storage account, so a delete against the + // configured account cannot be verified and would falsely report success if the store was repointed. + Mock store = new(); + TestLogger logger = new(); + DeleteExternalBlobActivity activity = new(store.Object, logger); + + // Act + BlobPurgeOutcome outcome = await activity.RunAsync(null!, "blob:v1:payloads:abc123"); + + // Assert - quarantined by the gate, and the store's DeleteAsync was never invoked. + outcome.Disposition.Should().Be(LargePayloadPurgeDisposition.Quarantined); + logger.Logs.Should().ContainSingle(l => l.Message.Contains("LegacyV1Token")); + store.Verify(s => s.DeleteAsync(It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact] + public async Task RunAsync_UnknownTokenVersion_RetriesWithoutCallingStore() + { + // Arrange - an unrecognized prefix most likely came from a newer SDK, which recovers after an upgrade. + Mock store = new(); + TestLogger logger = new(); + DeleteExternalBlobActivity activity = new(store.Object, logger); + + // Act + BlobPurgeOutcome outcome = await activity.RunAsync(null!, "blob:v9:https://acct.blob.core.windows.net/c/x"); + + // Assert - retried, NOT quarantined: quarantine is terminal and requires an operator to unwind, while a + // deferral only leaves the row idle and visible. This branch and the token branches that quarantine are + // told apart only by cause, so the disposition is asserted deliberately: folding them together would + // strand rows that an SDK upgrade would have resolved. + outcome.Disposition.Should().Be(LargePayloadPurgeDisposition.Retry); + logger.Logs.Should().ContainSingle(l => l.Message.Contains("UnsupportedTokenVersion")); + store.Verify(s => s.DeleteAsync(It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact] + public async Task RunAsync_MalformedV2Token_Quarantines() + { + // Arrange - a recognized v2 prefix whose body does not parse; the store signals that with + // ArgumentException. Retrying can never fix a body the SDK itself produced malformed. + StubPayloadStore store = new(new ArgumentException("Invalid token")); + TestLogger logger = new(); + DeleteExternalBlobActivity activity = new(store, logger); + + // Act + BlobPurgeOutcome outcome = await activity.RunAsync(null!, "blob:v2:not-a-uri"); + + // Assert - contrast with the unknown-prefix case above, which is retried. + outcome.Disposition.Should().Be(LargePayloadPurgeDisposition.Quarantined); + logger.Logs.Should().ContainSingle(l => l.Message.Contains("MalformedToken")); + } + + [Fact] + public async Task RunAsync_V2Token_CallsStoreAndReportsDeleted() + { + // 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())) + .ReturnsAsync(PayloadDeleteOutcome.Deleted); + TestLogger logger = new(); + DeleteExternalBlobActivity activity = new(store.Object, logger); + + // Act + BlobPurgeOutcome outcome = await activity.RunAsync(null!, V2Token); + + // Assert - an ordinary success is silent; any log here would mean a failure branch was taken. + outcome.Disposition.Should().Be(LargePayloadPurgeDisposition.Deleted); + logger.Logs.Should().BeEmpty(); + store.Verify(s => s.DeleteAsync(It.IsAny(), It.IsAny()), Times.Once); + } + + [Fact] + public async Task RunAsync_WhenBlobAlreadyAbsent_ReportsDeleted() + { + // Arrange - deletion is idempotent, so a blob a previous attempt already removed is not a failure. + Mock store = new(); + store.Setup(s => s.DeleteAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(PayloadDeleteOutcome.AlreadyAbsent); + TestLogger logger = new(); + DeleteExternalBlobActivity activity = new(store.Object, logger); + + // Act + BlobPurgeOutcome outcome = await activity.RunAsync(null!, V2Token); + + // Assert + outcome.Disposition.Should().Be(LargePayloadPurgeDisposition.Deleted); + logger.Logs.Should().BeEmpty(); + } + + [Fact] + public async Task RunAsync_WhenBlobNotStoreOwned_ResolvesTombstoneWithoutDeleting() + { + // Arrange - the blob exists but carries no ownership marker, so the store left it untouched. + Mock store = new(); + store.Setup(s => s.DeleteAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(PayloadDeleteOutcome.NotStoreOwned); + TestLogger logger = new(); + DeleteExternalBlobActivity activity = new(store.Object, logger); + + // Act + BlobPurgeOutcome outcome = await activity.RunAsync(null!, V2Token); + + // Assert - the tombstone is still resolved: a blob the store does not own is not the store's to delete, + // and re-serving the row forever would never make it deletable. The reported result is now identical to + // an ordinary delete, so the log is the only thing that records the difference. + outcome.Disposition.Should().Be(LargePayloadPurgeDisposition.Deleted); + logger.Logs.Should().ContainSingle(l => l.Message.Contains("ownership marker")); + } + + [Fact] + public async Task RunAsync_WhenStoreDoesNotSupportDelete_RetriesToPreserveTombstone() + { + // Arrange - a store that cannot delete (the base PayloadStore.DeleteAsync throws NotSupportedException). + StubPayloadStore store = new(new NotSupportedException()); + TestLogger logger = new(); + DeleteExternalBlobActivity activity = new(store, logger); + + // Act + BlobPurgeOutcome outcome = await activity.RunAsync(null!, V2Token); + + // Assert - retried (tombstone preserved): resolving it would destroy the backend's cleanup ledger while + // the blob survives. + outcome.Disposition.Should().Be(LargePayloadPurgeDisposition.Retry); + logger.Logs.Should().ContainSingle(l => l.Message.Contains("StoreCannotDelete")); + } + + [Fact] + public async Task RunAsync_WhenDeleteTimesOut_RetriesAsTransient() + { + // Arrange - a non-Azure exception (timeout / network failure) must not drop a blob on doubt. + StubPayloadStore store = new(new TimeoutException()); + TestLogger logger = new(); + DeleteExternalBlobActivity activity = new(store, logger); + + // Act + BlobPurgeOutcome outcome = await activity.RunAsync(null!, V2Token); + + // Assert - storage reported no code here, so the exception's type name is what identifies the failure. + outcome.Disposition.Should().Be(LargePayloadPurgeDisposition.Retry); + logger.Logs.Should().ContainSingle(l => l.Message.Contains("UnexpectedFailure:TimeoutException")); + } + + 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.FromResult(PayloadDeleteOutcome.Deleted) + : 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/BlobPayloadStoreTests.cs b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs index c03ee76b..4a2111b5 100644 --- a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs +++ b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs @@ -373,6 +373,48 @@ public async Task UploadAsync_StaleContainerNotFoundFailure_DoesNotOverwriteNewe createCalls.Should().Be(2); } + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task UploadAsync_WritesOwnershipMarkerOnBothWritePaths(bool compressionEnabled) + { + // Arrange - the marker is what lets auto-purge prove the store wrote a blob before deleting it, so it + // must be written on the compressed and the uncompressed path alike. It rides the existing + // BlobOpenWriteOptions, so it costs no extra request. + BlobOpenWriteOptions? capturedOptions = null; + Mock containerClientMock = new(); + containerClientMock.Setup(c => c.Name).Returns("test-container"); + containerClientMock + .Setup(c => c.GetBlobClient(It.IsAny())) + .Returns(() => + { + Mock blobClientMock = new(); + blobClientMock.SetupGet(b => b.Uri).Returns( + new Uri("https://testaccount.blob.core.windows.net/test-container/payload")); + blobClientMock + .Setup(b => b.OpenWriteAsync( + It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync((bool _, BlobOpenWriteOptions options, CancellationToken _) => + { + capturedOptions = options; + return new MemoryStream(); + }); + return blobClientMock.Object; + }); + + LargePayloadStorageOptions options = new() { CompressionEnabled = compressionEnabled }; + BlobPayloadStore store = new(options, containerClientMock.Object); + + // Act + await store.UploadAsync("payload", CancellationToken.None); + + // Assert + capturedOptions.Should().NotBeNull(); + capturedOptions!.Metadata.Should().NotBeNull(); + capturedOptions.Metadata.Should().Contain( + BlobPayloadStore.OwnershipMarkerName, BlobPayloadStore.OwnershipMarkerValue); + } + static Mock CreateContainerClientMock() { Mock containerClientMock = new(); diff --git a/test/Extensions/AzureBlobPayloads.Tests/DependencyInjection/UseExternalizedPayloadsTests.cs b/test/Extensions/AzureBlobPayloads.Tests/DependencyInjection/UseExternalizedPayloadsTests.cs new file mode 100644 index 00000000..7ab88c15 --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/DependencyInjection/UseExternalizedPayloadsTests.cs @@ -0,0 +1,157 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using FluentAssertions; +using Grpc.Core; +using Microsoft.DurableTask.AzureBlobPayloads; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.Grpc; +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 clientProvider = services.BuildServiceProvider(); + + // Assert - the store resolves without throwing and is the blob-backed implementation. + PayloadStore store = clientProvider.GetRequiredService(); + store.Should().BeOfType(); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public void UseExternalizedPayloads_Worker_SendsResolvedAutoPurgeValueOnTheHandshake(bool autoPurge) + { + // Arrange - the backend only tombstones payloads for a task hub whose workers opted in, so the resolved + // AutoPurge value has to ride the GetWorkItems handshake. A worker that never calls this extension sends + // nothing at all, which the backend reads as "no opinion". + ServiceCollection services = new(); + Mock builder = new(); + builder.Setup(b => b.Services).Returns(services); + builder.Setup(b => b.Name).Returns(string.Empty); + services.AddOptions(string.Empty) + .Configure(o => o.CallInvoker = Mock.Of()); + + // Act + builder.Object.UseExternalizedPayloads(options => + { + options.ConnectionString = "UseDevelopmentStorage=true"; + options.AutoPurge = autoPurge; + }); + + using ServiceProvider provider = services.BuildServiceProvider(); + GrpcDurableTaskWorkerOptions grpcOptions = + provider.GetRequiredService>().Get(string.Empty); + + // Assert - an explicit true/false, never left absent, because calling this extension IS the choice. + grpcOptions.LargePayloadAutoPurgeEnabled.Should().Be(autoPurge); + } +} 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..30c7cee3 --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/PayloadStore/BlobPayloadStoreDeleteTests.cs @@ -0,0 +1,253 @@ +// 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, the +/// self-describing v2 token resolution (same account, cross-account with identity, cross-account without), and +/// the ownership marker that gates every delete. +/// +public class BlobPayloadStoreDeleteTests +{ + const string ContainerName = "payloads"; + const string ConfiguredAccountUrl = "https://myaccount.blob.core.windows.net"; + + static readonly ETag KnownETag = new("\"0x8DTEST\""); + + 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; + } + + /// + /// Creates a blob that exists and carries this store's ownership marker, which is the ordinary case for a + /// payload the store itself uploaded. + /// + static Mock CreateBlob(bool existed) => CreateBlob(existed, owned: true); + + static Mock CreateBlob(bool existed, bool owned) + { + Mock blob = new(); + + if (existed) + { + Dictionary metadata = owned + ? new() { [BlobPayloadStore.OwnershipMarkerName] = BlobPayloadStore.OwnershipMarkerValue } + : new() { ["customer-tag"] = "not-ours" }; + + blob + .Setup(b => b.GetPropertiesAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(Response.FromValue( + BlobsModelFactory.BlobProperties(metadata: metadata, eTag: KnownETag), Mock.Of())); + } + else + { + blob + .Setup(b => b.GetPropertiesAsync(It.IsAny(), It.IsAny())) + .ThrowsAsync(new RequestFailedException(404, "not found", "BlobNotFound", null)); + } + + 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 + PayloadDeleteOutcome outcome = await store.DeleteAsync( + $"blob:v1:{ContainerName}:abc123", CancellationToken.None); + + // Assert + outcome.Should().Be(PayloadDeleteOutcome.Deleted); + container.Verify(c => c.GetBlobClient("abc123"), Times.Once); + blob.Verify( + b => b.DeleteIfExistsAsync( + DeleteSnapshotsOption.IncludeSnapshots, + It.Is(c => c.IfMatch == KnownETag), + It.IsAny()), + Times.Once); + } + + [Fact] + public async Task DeleteAsync_MissingBlob_IsIdempotentAndReportsAlreadyAbsent() + { + // 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) + PayloadDeleteOutcome outcome = await store.DeleteAsync( + $"blob:v1:{ContainerName}:missing", CancellationToken.None); + + // Assert - the ownership probe already proved absence, so no delete request is needed. + outcome.Should().Be(PayloadDeleteOutcome.AlreadyAbsent); + blob.Verify( + b => b.DeleteIfExistsAsync( + It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + } + + [Fact] + public async Task DeleteAsync_BlobWithoutOwnershipMarker_LeavesBlobUntouched() + { + // Arrange - a blob whose token text matches the v2 grammar but which this store never wrote (for + // example a customer dataset referenced by URL, or a payload written before the marker shipped). + Mock blob = CreateBlob(existed: true, owned: false); + Mock container = CreateContainer(blob, "abc123"); + BlobPayloadStore store = new(new LargePayloadStorageOptions(), container.Object); + + // Act + PayloadDeleteOutcome outcome = await store.DeleteAsync( + $"blob:v2:{ConfiguredAccountUrl}/{ContainerName}/abc123", CancellationToken.None); + + // Assert - reported distinctly and, critically, never deleted. + outcome.Should().Be(PayloadDeleteOutcome.NotStoreOwned); + blob.Verify( + b => b.DeleteIfExistsAsync( + It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + } + + [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 + PayloadDeleteOutcome outcome = await store.DeleteAsync( + $"blob:v2:{ConfiguredAccountUrl}/{ContainerName}/abc123", CancellationToken.None); + + // Assert + outcome.Should().Be(PayloadDeleteOutcome.Deleted); + container.Verify(c => c.GetBlobClient("abc123"), Times.Once); + blob.Verify( + b => b.DeleteIfExistsAsync( + DeleteSnapshotsOption.IncludeSnapshots, + It.Is(c => c.IfMatch == KnownETag), + 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 + { + } + } +}