diff --git a/Microsoft.DurableTask.sln b/Microsoft.DurableTask.sln index 3c6d4539..bf471507 100644 --- a/Microsoft.DurableTask.sln +++ b/Microsoft.DurableTask.sln @@ -1,4 +1,4 @@ - + Microsoft Visual Studio Solution File, Format Version 12.00 # Visual Studio Version 17 VisualStudioVersion = 17.3.32901.215 @@ -145,6 +145,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "AzureManaged", "AzureManage EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Grpc", "Grpc", "{3B8F957E-7773-4C0C-ACD7-91A1591D9312}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "AzureBlobPayloads.Tests", "test\AzureBlobPayloads.Tests\AzureBlobPayloads.Tests.csproj", "{531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -839,6 +841,18 @@ Global {C1995163-1DCE-405D-BE82-8B4B2584893E}.Release|x64.Build.0 = Release|Any CPU {C1995163-1DCE-405D-BE82-8B4B2584893E}.Release|x86.ActiveCfg = Release|Any CPU {C1995163-1DCE-405D-BE82-8B4B2584893E}.Release|x86.Build.0 = Release|Any CPU + {531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Debug|Any CPU.Build.0 = Debug|Any CPU + {531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Debug|x64.ActiveCfg = Debug|Any CPU + {531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Debug|x64.Build.0 = Debug|Any CPU + {531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Debug|x86.ActiveCfg = Debug|Any CPU + {531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Debug|x86.Build.0 = Debug|Any CPU + {531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Release|Any CPU.ActiveCfg = Release|Any CPU + {531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Release|Any CPU.Build.0 = Release|Any CPU + {531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Release|x64.ActiveCfg = Release|Any CPU + {531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Release|x64.Build.0 = Release|Any CPU + {531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Release|x86.ActiveCfg = Release|Any CPU + {531B29A0-B2AD-43E2-8F65-0BA85C82C4B5}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -911,6 +925,7 @@ Global {C1995163-1DCE-405D-BE82-8B4B2584893E} = {9686B8F9-2644-6C9B-E567-55B0471E4584} {53193780-CD18-2643-6953-C26F59EAEDF5} = {5B448FF6-EC42-491D-A22E-1DC8B618E6D5} {3B8F957E-7773-4C0C-ACD7-91A1591D9312} = {5B448FF6-EC42-491D-A22E-1DC8B618E6D5} + {531B29A0-B2AD-43E2-8F65-0BA85C82C4B5} = {E5637F81-2FB9-4CD7-900D-455363B142A7} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {AB41CB55-35EA-4986-A522-387AB3402E71} diff --git a/src/Client/Core/DurableTaskClient.cs b/src/Client/Core/DurableTaskClient.cs index 03303800..208e6441 100644 --- a/src/Client/Core/DurableTaskClient.cs +++ b/src/Client/Core/DurableTaskClient.cs @@ -549,6 +549,30 @@ public virtual Task> ListInstanceIdsAsync( $"{this.GetType()} does not support listing orchestration instance IDs filtered by completed time."); } + /// + /// Gets a batch of tombstoned (soft-deleted) externalized payloads whose backing blobs should be deleted + /// by a credentialed caller before the backend hard-deletes the rows. + /// + /// The maximum number of tombstoned payloads to request. + /// The cancellation token. + /// The batch of tombstoned payloads whose blobs should be deleted. + /// Thrown if this implementation does not support the operation. + public virtual Task> GetTombstonedPayloadsAsync( + int limit, CancellationToken cancellation = default) + => throw new NotSupportedException($"{this.GetType()} does not support retrieving tombstoned payloads."); + + /// + /// Acknowledges tombstoned payloads whose backing blobs have been deleted so the backend can hard-delete + /// the corresponding rows. + /// + /// The payloads whose blobs have been deleted. + /// The cancellation token. + /// A task that completes when the acknowledgement has been sent. + /// Thrown if this implementation does not support the operation. + public virtual Task AckPurgedPayloadsAsync( + IEnumerable acks, CancellationToken cancellation = default) + => throw new NotSupportedException($"{this.GetType()} does not support acknowledging purged payloads."); + // TODO: Create task hub // TODO: Delete task hub diff --git a/src/Client/Core/PayloadPurgeAckDto.cs b/src/Client/Core/PayloadPurgeAckDto.cs new file mode 100644 index 00000000..01db0f21 --- /dev/null +++ b/src/Client/Core/PayloadPurgeAckDto.cs @@ -0,0 +1,13 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.Client; + +/// +/// Serializable acknowledgement that the worker has deleted the blob for a tombstoned payload, so the +/// backend can hard-delete the soft-deleted row. Mirrors the PayloadPurgeAck protobuf message. +/// +/// The backend partition that owns the payload row. +/// The orchestration instance key the payload belongs to. +/// The backend identifier of the soft-deleted payload row. +public sealed record PayloadPurgeAckDto(int PartitionId, long InstanceKey, long PayloadId); diff --git a/src/Client/Core/TombstonedPayloadDto.cs b/src/Client/Core/TombstonedPayloadDto.cs new file mode 100644 index 00000000..69308d47 --- /dev/null +++ b/src/Client/Core/TombstonedPayloadDto.cs @@ -0,0 +1,15 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.Client; + +/// +/// Serializable representation of a tombstoned payload the backend has soft-deleted and whose blob the +/// worker should delete. Mirrors the TombstonedPayload protobuf message but is safe to pass through +/// the orchestration/activity boundary. +/// +/// The backend partition that owns the payload row. +/// The orchestration instance key the payload belongs to. +/// The backend identifier of the soft-deleted payload row. +/// The externalized payload token whose backing blob should be deleted. +public sealed record TombstonedPayloadDto(int PartitionId, long InstanceKey, long PayloadId, string Token); diff --git a/src/Client/Grpc/GrpcDurableTaskClient.cs b/src/Client/Grpc/GrpcDurableTaskClient.cs index 23350d4c..822f7238 100644 --- a/src/Client/Grpc/GrpcDurableTaskClient.cs +++ b/src/Client/Grpc/GrpcDurableTaskClient.cs @@ -624,6 +624,49 @@ public override async Task> GetOrchestrationHistoryAsync( } } + /// + public override async Task> GetTombstonedPayloadsAsync( + int limit, CancellationToken cancellation = default) + { + P.GetTombstonedPayloadsResponse response = await this.sidecarClient.GetTombstonedPayloadsAsync( + new P.GetTombstonedPayloadsRequest { Limit = limit }, + cancellationToken: cancellation); + + List result = new(response.Payloads.Count); + foreach (P.TombstonedPayload payload in response.Payloads) + { + result.Add(new TombstonedPayloadDto( + payload.PartitionId, payload.InstanceKey, payload.PayloadId, payload.Token)); + } + + return result; + } + + /// + public override async Task AckPurgedPayloadsAsync( + IEnumerable acks, CancellationToken cancellation = default) + { + Check.NotNull(acks); + + P.AckPurgedPayloadsRequest request = new(); + foreach (PayloadPurgeAckDto ack in acks) + { + request.Acks.Add(new P.PayloadPurgeAck + { + PartitionId = ack.PartitionId, + InstanceKey = ack.InstanceKey, + PayloadId = ack.PayloadId, + }); + } + + if (request.Acks.Count == 0) + { + return; + } + + await this.sidecarClient.AckPurgedPayloadsAsync(request, cancellationToken: cancellation); + } + static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, ILogger logger, out CallInvoker callInvoker) { Func>? recreator = options.Internal.ChannelRecreator; diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/AckPurgedPayloadsActivity.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/AckPurgedPayloadsActivity.cs new file mode 100644 index 00000000..637fa5ec --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/AckPurgedPayloadsActivity.cs @@ -0,0 +1,36 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Client; +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Activity that acknowledges to the backend the payloads whose blobs the worker has deleted, so the backend +/// can hard-delete the soft-deleted rows. +/// +/// The Durable Task client used to acknowledge purged payloads to the backend. +/// The logger instance. +[DurableTask] +public class AckPurgedPayloadsActivity( + DurableTaskClient client, + ILogger logger) + : TaskActivity, object?> +{ + readonly DurableTaskClient client = Check.NotNull(client); + readonly ILogger logger = Check.NotNull(logger); + + /// + public override async Task RunAsync(TaskActivityContext context, List input) + { + if (input is null || input.Count == 0) + { + return null; + } + + await this.client.AckPurgedPayloadsAsync(input, CancellationToken.None); + this.logger.BlobPurgeAckedPayloads(input.Count); + return null; + } +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs new file mode 100644 index 00000000..0018cb7c --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs @@ -0,0 +1,42 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Activity that deletes a single externalized payload blob given its token. Deletion is idempotent, so +/// re-delivered tokens and concurrent workers are safe. On failure the payload is left tombstoned so a later +/// purge cycle can retry it. +/// +/// 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)); + + try + { + await this.store.DeleteAsync(input, CancellationToken.None); + return true; + } + catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException) + { + // Leave the payload tombstoned so the backend re-streams it on a later cycle; a single bad token + // must not fail the whole batch. + this.logger.BlobPurgeDeleteFailed(ex, input); + return false; + } + } +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs new file mode 100644 index 00000000..6086efe2 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs @@ -0,0 +1,32 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Client; +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Activity that fetches a batch of tombstoned payloads from the backend for the auto-purge job to delete. +/// +/// The Durable Task client used to query the backend for tombstoned payloads. +/// The logger instance. +[DurableTask] +public class GetTombstonedPayloadsActivity( + DurableTaskClient client, + ILogger logger) + : TaskActivity> +{ + readonly DurableTaskClient client = Check.NotNull(client); + readonly ILogger logger = Check.NotNull(logger); + + /// + public override async Task> RunAsync(TaskActivityContext context, int input) + { + int limit = input > 0 ? input : 500; + List payloads = + await this.client.GetTombstonedPayloadsAsync(limit, CancellationToken.None); + this.logger.BlobPurgeFetchedTombstones(payloads.Count); + return payloads; + } +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs new file mode 100644 index 00000000..9652c1e3 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs @@ -0,0 +1,135 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +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 ensures the singleton blob payload auto-purge job exists when auto-purge +/// is enabled via . It never blocks host startup: it runs +/// on a background task and retries until the backend is reachable. The job is a whole-scheduler singleton, +/// so racing client processes simply no-op. +/// +sealed class BlobPurgeJobStarter : IHostedService +{ + static readonly TimeSpan RetryDelay = TimeSpan.FromSeconds(10); + + readonly DurableTaskClient client; + readonly IOptionsMonitor options; + readonly string builderName; + readonly ILogger logger; + readonly EntityInstanceId entityId = new(nameof(BlobPurgeJob), BlobPurgeConstants.JobId); + + CancellationTokenSource? cts; + Task? ensureTask; + + public BlobPurgeJobStarter( + DurableTaskClient client, + IOptionsMonitor options, + string builderName, + ILogger logger) + { + this.client = Check.NotNull(client); + this.options = Check.NotNull(options); + this.builderName = Check.NotNull(builderName); + this.logger = Check.NotNull(logger); + } + + /// + public Task StartAsync(CancellationToken cancellationToken) + { + LargePayloadStorageOptions opts = this.options.Get(this.builderName); + if (!opts.AutoPurge) + { + this.logger.BlobPurgeDisabled(); + return Task.CompletedTask; + } + + int batchSize = opts.PayloadPurgeBatchSize > 0 ? opts.PayloadPurgeBatchSize : 500; + + // Do not block host startup; ensure the job on a background task with basic retry until the backend + // is reachable. + this.cts = new CancellationTokenSource(); + this.ensureTask = Task.Run(() => this.EnsureJobAsync(batchSize, this.cts.Token), CancellationToken.None); + return Task.CompletedTask; + } + + /// + public async Task StopAsync(CancellationToken cancellationToken) + { + this.cts?.Cancel(); + + Task? pending = this.ensureTask; + if (pending is not null) + { + // The ensure loop observes cancellation and returns promptly; swallow any faulted/cancelled result. + await Task.WhenAny(pending, Task.Delay(Timeout.Infinite, cancellationToken)).ConfigureAwait(false); + } + } + + async Task EnsureJobAsync(int batchSize, CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested) + { + try + { + if (await this.IsJobActiveAsync(cancellationToken)) + { + return; + } + + BlobPurgeJobOperationRequest request = new( + this.entityId, + nameof(BlobPurgeJob.Create), + new BlobPurgeJobCreationOptions(batchSize)); + + await this.client.ScheduleNewOrchestrationInstanceAsync( + new TaskName(nameof(ExecuteBlobPurgeJobOperationOrchestrator)), + request, + cancellationToken); + + this.logger.BlobPurgeJobEnsured(); + return; + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + return; + } + catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException) + { + this.logger.BlobPurgeStarterRetry(ex); + try + { + await Task.Delay(RetryDelay, cancellationToken); + } + catch (OperationCanceledException) + { + return; + } + } + } + } + + async Task IsJobActiveAsync(CancellationToken cancellationToken) + { + try + { + EntityMetadata? metadata = + await this.client.Entities.GetEntityAsync( + this.entityId, cancellation: cancellationToken); + return metadata is not null && metadata.State.Status == BlobPurgeJobStatus.Active; + } + catch (NotSupportedException) + { + // The entity-query API is unavailable on this client; fall back to scheduling the idempotent + // Create, which no-ops if the job is already active. + return false; + } + } +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs new file mode 100644 index 00000000..eafee991 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs @@ -0,0 +1,28 @@ +// 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 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..8f08b867 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs @@ -0,0 +1,84 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Entities; +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Durable entity that manages the lifecycle of the singleton blob payload auto-purge job. +/// +/// The logger instance. +class BlobPurgeJob(ILogger logger) : TaskEntity +{ + /// + /// Creates (or reactivates) the auto-purge job. Because the job is a whole-scheduler singleton, this is + /// intentionally a no-op when the job is already so that extra + /// client processes racing to create it do not disturb the running job. + /// + /// The entity context. + /// The job creation options. + public void Create(TaskEntityContext context, BlobPurgeJobCreationOptions creationOptions) + { + Check.NotNull(creationOptions, nameof(creationOptions)); + + if (this.State.Status == BlobPurgeJobStatus.Active) + { + logger.BlobPurgeJobAlreadyRunning(context.Id.Key); + return; + } + + this.State.Status = BlobPurgeJobStatus.Active; + this.State.PurgeBatchSize = creationOptions.PurgeBatchSize > 0 ? creationOptions.PurgeBatchSize : 500; + this.State.CreatedAt ??= DateTimeOffset.UtcNow; + this.State.LastModifiedAt = DateTimeOffset.UtcNow; + this.State.LastError = null; + + logger.BlobPurgeJobCreated(context.Id.Key); + + // Signal Run to start the perpetual purge orchestrator. + context.SignalEntity(context.Id, nameof(this.Run)); + } + + /// + /// Starts the purge orchestrator if the job is active. Uses a fixed orchestrator instance ID so only one + /// orchestrator ever runs for the singleton job. + /// + /// The entity context. + public void Run(TaskEntityContext context) + { + if (this.State.Status != BlobPurgeJobStatus.Active) + { + return; + } + + string instanceId = BlobPurgeConstants.GetOrchestratorInstanceId(context.Id.Key); + StartOrchestrationOptions startOrchestrationOptions = new(instanceId); + + context.ScheduleNewOrchestration( + new TaskName(nameof(BlobPurgeJobOrchestrator)), + new BlobPurgeJobRunRequest(context.Id, this.State.PurgeBatchSize), + startOrchestrationOptions); + + this.State.LastModifiedAt = DateTimeOffset.UtcNow; + } + + /// + /// Records progress after a purge cycle completes. + /// + /// The entity context. + /// The number of blobs purged in the cycle. + public void RecordPurged(TaskEntityContext context, long purgedCount) + { + this.State.PurgedCount += purgedCount; + this.State.LastModifiedAt = DateTimeOffset.UtcNow; + } + + /// + /// Gets the current state of the auto-purge job. + /// + /// The entity context. + /// The current job state. + public BlobPurgeJobState Get(TaskEntityContext context) => this.State; +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs new file mode 100644 index 00000000..6e4ebed8 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs @@ -0,0 +1,39 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Log messages for the Azure Blob externalized-payload auto-purge job. +/// +static partial class Logs +{ + [LoggerMessage(EventId = 810, Level = LogLevel.Information, Message = "Blob payload auto-purge job '{jobId}' created.")] + public static partial void BlobPurgeJobCreated(this ILogger logger, string? jobId); + + [LoggerMessage(EventId = 811, Level = LogLevel.Information, Message = "Blob payload auto-purge job '{jobId}' is already running; ignoring the create request.")] + public static partial void BlobPurgeJobAlreadyRunning(this ILogger logger, string? jobId); + + [LoggerMessage(EventId = 812, Level = LogLevel.Information, Message = "Blob payload auto-purge orchestrator for job '{jobId}' stopping; job status is {status}.")] + public static partial void BlobPurgeJobOrchestratorStopping(this ILogger logger, string? jobId, string status); + + [LoggerMessage(EventId = 813, Level = LogLevel.Warning, Message = "Failed to delete externalized payload blob for token '{token}'; leaving it tombstoned for a later purge cycle.")] + public static partial void BlobPurgeDeleteFailed(this ILogger logger, Exception exception, string token); + + [LoggerMessage(EventId = 814, Level = LogLevel.Debug, Message = "Blob payload auto-purge fetched {count} tombstoned payload(s) from the backend.")] + public static partial void BlobPurgeFetchedTombstones(this ILogger logger, int count); + + [LoggerMessage(EventId = 815, Level = LogLevel.Debug, Message = "Blob payload auto-purge acknowledged {count} purged payload(s) to the backend.")] + public static partial void BlobPurgeAckedPayloads(this ILogger logger, int count); + + [LoggerMessage(EventId = 816, Level = LogLevel.Information, Message = "Blob payload auto-purge is disabled; the singleton purge job will not be started.")] + public static partial void BlobPurgeDisabled(this ILogger logger); + + [LoggerMessage(EventId = 817, Level = LogLevel.Information, Message = "Blob payload auto-purge singleton job ensured.")] + public static partial void BlobPurgeJobEnsured(this ILogger logger); + + [LoggerMessage(EventId = 818, Level = LogLevel.Warning, Message = "Blob payload auto-purge starter could not ensure the singleton job; retrying.")] + public static partial void BlobPurgeStarterRetry(this ILogger logger, Exception exception); +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobCreationOptions.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobCreationOptions.cs new file mode 100644 index 00000000..e49e423c --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobCreationOptions.cs @@ -0,0 +1,12 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Options used to create the singleton blob payload auto-purge job. +/// +/// +/// The maximum number of tombstoned payloads to request from the backend per cycle. +/// +public sealed record BlobPurgeJobCreationOptions(int PurgeBatchSize); diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobState.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobState.cs new file mode 100644 index 00000000..8bd4cdd5 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobState.cs @@ -0,0 +1,40 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// State for the singleton blob payload auto-purge job, stored in the entity. +/// +public sealed class BlobPurgeJobState +{ + /// + /// Gets or sets the current status of the auto-purge job. + /// + public BlobPurgeJobStatus Status { get; set; } + + /// + /// Gets or sets the time when the job was first created. + /// + public DateTimeOffset? CreatedAt { get; set; } + + /// + /// Gets or sets the time when the job state was last modified. + /// + public DateTimeOffset? LastModifiedAt { get; set; } + + /// + /// Gets or sets the total number of payload blobs the job has purged. + /// + public long PurgedCount { get; set; } + + /// + /// Gets or sets the last error message, if any. + /// + public string? LastError { get; set; } + + /// + /// Gets or sets the maximum number of tombstoned payloads requested from the backend per cycle. + /// + public int PurgeBatchSize { get; set; } +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobStatus.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobStatus.cs new file mode 100644 index 00000000..917b8d1b --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobStatus.cs @@ -0,0 +1,26 @@ +// 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 the default status of a freshly initialized entity, so it is kept + /// as the zero value to avoid a brand-new entity accidentally appearing active. + /// + Stopped, + + /// + /// The job is active and draining tombstoned payloads from the backend. + /// + Active, + + /// + /// The job has failed. + /// + Failed, +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs new file mode 100644 index 00000000..2b7ba6f2 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs @@ -0,0 +1,145 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Entities; +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Orchestrator input describing the purge job to run. +/// +/// The entity ID of the owning . +/// The maximum number of tombstoned payloads to request per cycle. +/// The number of cycles processed since the last continue-as-new. +public sealed record BlobPurgeJobRunRequest( + EntityInstanceId JobEntityId, int PurgeBatchSize, int ProcessedCycles = 0); + +/// +/// Perpetual orchestrator that drains tombstoned payloads from the backend, deletes their blobs with capped +/// parallelism, and acknowledges the successful deletions so the backend can hard-delete the rows. It idles +/// on a timer when there is nothing to purge and continues-as-new periodically to keep its history small. +/// +[DurableTask] +public class BlobPurgeJobOrchestrator : TaskOrchestrator +{ + const int ContinueAsNewFrequency = 5; + const int MaxParallelDeletes = 32; + const int DefaultPurgeBatchSize = 500; + static readonly TimeSpan IdleDelay = 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 > 0 ? input.PurgeBatchSize : DefaultPurgeBatchSize; + int processedCycles = input.ProcessedCycles; + + while (true) + { + processedCycles++; + if (processedCycles > ContinueAsNewFrequency) + { + context.ContinueAsNew(new BlobPurgeJobRunRequest(input.JobEntityId, batchSize, ProcessedCycles: 0)); + return null!; + } + + // Stop cleanly if the job has been stopped or removed. + BlobPurgeJobState? state = await context.Entities.CallEntityAsync( + input.JobEntityId, nameof(BlobPurgeJob.Get), null); + + if (state is null || state.Status != BlobPurgeJobStatus.Active) + { + logger.BlobPurgeJobOrchestratorStopping(jobId, state?.Status.ToString() ?? "null"); + return null; + } + + List tombstones = await context.CallActivityAsync>( + nameof(GetTombstonedPayloadsActivity), + batchSize, + new TaskOptions(PurgeActivityRetryPolicy)); + + if (tombstones is null || tombstones.Count == 0) + { + // Nothing to purge right now: block on a timer (push-free idle) then check again. + await context.CreateTimer(IdleDelay, default); + continue; + } + + List deleted = await this.DeleteBatchAsync(context, tombstones); + + if (deleted.Count > 0) + { + await context.CallActivityAsync( + nameof(AckPurgedPayloadsActivity), + deleted, + new TaskOptions(PurgeActivityRetryPolicy)); + + await context.Entities.CallEntityAsync( + input.JobEntityId, nameof(BlobPurgeJob.RecordPurged), (long)deleted.Count); + } + } + } + + async Task> DeleteBatchAsync( + TaskOrchestrationContext context, List tombstones) + { + List deleted = new(tombstones.Count); + List> tasks = new(); + + foreach (TombstonedPayloadDto tombstone in tombstones) + { + tasks.Add(this.DeleteOneAsync(context, tombstone)); + + if (tasks.Count >= MaxParallelDeletes) + { + await DrainAsync(tasks, deleted); + tasks.Clear(); + } + } + + if (tasks.Count > 0) + { + await DrainAsync(tasks, deleted); + } + + return deleted; + } + + static async Task DrainAsync(List> tasks, List deleted) + { + DeleteOutcome[] outcomes = await Task.WhenAll(tasks); + foreach (DeleteOutcome outcome in outcomes) + { + // Only acknowledge blobs that were actually deleted; failed tokens stay tombstoned to retry. + if (outcome.Deleted) + { + deleted.Add(outcome.Ack); + } + } + } + + async Task DeleteOneAsync(TaskOrchestrationContext context, TombstonedPayloadDto tombstone) + { + bool deleted = await context.CallActivityAsync( + nameof(DeleteExternalBlobActivity), + tombstone.Token, + new TaskOptions(PurgeActivityRetryPolicy)); + + return new DeleteOutcome( + deleted, + new PayloadPurgeAckDto(tombstone.PartitionId, tombstone.InstanceKey, tombstone.PayloadId)); + } + + readonly record struct DeleteOutcome(bool Deleted, PayloadPurgeAckDto Ack); +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/ExecuteBlobPurgeJobOperationOrchestrator.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/ExecuteBlobPurgeJobOperationOrchestrator.cs new file mode 100644 index 00000000..e8f0c66b --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/ExecuteBlobPurgeJobOperationOrchestrator.cs @@ -0,0 +1,30 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Entities; + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// Orchestrator that executes a single operation on a blob purge job entity and returns the result. Used as +/// a client-to-entity bridge so clients can drive the entity through the orchestration surface. +/// +[DurableTask] +public class ExecuteBlobPurgeJobOperationOrchestrator + : TaskOrchestrator +{ + /// + public override async Task RunAsync( + TaskOrchestrationContext context, BlobPurgeJobOperationRequest input) + { + return await context.Entities.CallEntityAsync(input.EntityId, input.OperationName, input.Input); + } +} + +/// +/// Request for executing a blob purge job entity operation. +/// +/// The ID of the entity to execute the operation on. +/// The name of the operation to execute. +/// Optional input for the operation. +public record BlobPurgeJobOperationRequest(EntityInstanceId EntityId, string OperationName, object? Input = null); diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs index 0817bcea..2557c6da 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs @@ -2,11 +2,14 @@ // 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.Hosting; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; namespace Microsoft.DurableTask; @@ -16,6 +19,29 @@ 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); + builder.Services.AddSingleton(sp => + { + LargePayloadStorageOptions opts = sp.GetRequiredService>().Get(builder.Name); + return new BlobPayloadStore(opts); + }); + + 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. @@ -56,6 +82,18 @@ static IDurableTaskClientBuilder UseExternalizedPayloadsCore(IDurableTaskClientB } }); + RegisterBlobPurgeJobStarter(builder); + return builder; } + + static void RegisterBlobPurgeJobStarter(IDurableTaskClientBuilder builder) + { + string builderName = builder.Name; + builder.Services.AddSingleton(sp => new BlobPurgeJobStarter( + 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..42ac0f0a 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs @@ -2,8 +2,7 @@ // 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; @@ -82,6 +81,19 @@ static IDurableTaskWorkerBuilder UseExternalizedPayloadsCore(IDurableTaskWorkerB opt.Capabilities.Add(P.WorkerCapability.LargePayloads); }); + // Register the entity/orchestrators/activities that run the singleton auto-purge job. These are + // ALWAYS registered (not gated on AutoPurge) so that a client-enabled job always has something to + // execute here. The purge activities fetch/ack via the injected DurableTaskClient. + builder.AddTasks(r => + { + r.AddEntity(); + r.AddOrchestrator(); + r.AddOrchestrator(); + r.AddActivity(); + r.AddActivity(); + r.AddActivity(); + }); + return builder; } } diff --git a/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs b/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs index 6abcbdf2..f53c9e72 100644 --- a/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs +++ b/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs @@ -115,4 +115,18 @@ 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). + /// + public bool AutoPurge { get; set; } + + /// + /// Gets or sets the maximum number of tombstoned payloads the auto-purge job requests from the backend + /// per cycle. Defaults to 500. Values less than or equal to zero are treated as the default. + /// + public int PayloadPurgeBatchSize { get; set; } = 500; } diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs index e01d2e74..1934c5d4 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -67,6 +67,18 @@ public BlobPayloadStore(LargePayloadStorageOptions options) this.containerClient = serviceClient.GetBlobContainerClient(options.ContainerName); } + /// + /// Initializes a new instance of the class using a caller-supplied + /// container client. Intended for unit testing so a mocked can be injected. + /// + /// The blob container client to use. + /// The options for the blob payload store. + internal BlobPayloadStore(BlobContainerClient containerClient, LargePayloadStorageOptions options) + { + this.containerClient = containerClient ?? throw new ArgumentNullException(nameof(containerClient)); + this.options = options ?? throw new ArgumentNullException(nameof(options)); + } + /// public override async Task UploadAsync(string payLoad, CancellationToken cancellationToken) { @@ -145,6 +157,25 @@ public override async Task DownloadAsync(string token, CancellationToken } } + /// + public override async Task DeleteAsync(string token, CancellationToken cancellationToken) + { + (string container, string name) = DecodeToken(token); + if (!string.Equals(container, this.containerClient.Name, StringComparison.Ordinal)) + { + throw new ArgumentException("Token container does not match configured container.", nameof(token)); + } + + BlobClient blob = this.containerClient.GetBlobClient(name); + + // Idempotent by design: DeleteIfExistsAsync returns false (rather than throwing) when the blob is + // already gone, so re-delivered tombstones and concurrent purges from multiple worker replicas are safe. + await blob.DeleteIfExistsAsync( + DeleteSnapshotsOption.IncludeSnapshots, + conditions: null, + cancellationToken); + } + /// public override bool IsKnownPayloadToken(string value) { diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs index b0fe6f80..0ae5ccb6 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs @@ -24,6 +24,22 @@ public abstract class PayloadStore /// Payload string. public abstract Task DownloadAsync(string token, CancellationToken cancellationToken); + /// + /// Deletes the payload referenced by the token. Implementations that support deletion must be + /// idempotent: deleting a payload that no longer exists is a no-op and must not throw. + /// + /// + /// The default implementation throws . Stores that externalize + /// payloads to deletable storage (for example Azure Blob Storage) should override it. It is declared + /// virtual rather than abstract so that adding it does not break existing external subclasses. + /// + /// The opaque reference token. + /// Cancellation token. + /// A task that completes when the payload has been deleted (or was already absent). + public virtual Task DeleteAsync(string token, CancellationToken cancellationToken) => + throw new NotSupportedException( + $"This {nameof(PayloadStore)} implementation does not support deleting payloads."); + /// /// Returns true if the specified value appears to be a token understood by this store. /// Implementations should not throw for unknown tokens. diff --git a/src/Grpc/orchestrator_service.proto b/src/Grpc/orchestrator_service.proto index 3d9194ac..af065941 100644 --- a/src/Grpc/orchestrator_service.proto +++ b/src/Grpc/orchestrator_service.proto @@ -823,6 +823,53 @@ service TaskHubSidecarService { // "Skip" graceful termination of orchestrations by immediately changing their status in storage to "terminated". // Note that a maximum of 500 orchestrations can be terminated at a time using this method. rpc SkipGracefulOrchestrationTerminations(SkipGracefulOrchestrationTerminationsRequest) returns (SkipGracefulOrchestrationTerminationsResponse); + + // Returns a batch of blob-externalized payload tombstones the backend has soft-deleted so the worker + // can delete the corresponding blobs from customer storage (the backend has no storage credentials of + // its own). The worker deletes each blob and then calls AckPurgedPayloads so the backend can + // hard-delete the soft-deleted rows. This is an opt-in, worker-driven pull model. + rpc GetTombstonedPayloads(GetTombstonedPayloadsRequest) returns (GetTombstonedPayloadsResponse); + + // Acknowledges that the worker has deleted the blobs for the identified payloads so the backend can + // hard-delete the soft-deleted rows. + rpc AckPurgedPayloads(AckPurgedPayloadsRequest) returns (AckPurgedPayloadsResponse); +} + +// Server -> client. Identifies a blob-externalized payload that the backend has soft-deleted and whose +// blob the worker should delete from customer storage. +message TombstonedPayload { + int32 partitionId = 1; + int64 instanceKey = 2; + int64 payloadId = 3; + // The externalized payload token (e.g. "blob:v1::") whose backing blob should be deleted. + string token = 4; +} + +// Client -> server. Acknowledges that the worker has deleted the blob for the identified payload so the +// backend can hard-delete the soft-deleted row. +message PayloadPurgeAck { + int32 partitionId = 1; + int64 instanceKey = 2; + int64 payloadId = 3; +} + +// Client -> server. Requests up to `limit` tombstoned payloads for the worker to delete. +message GetTombstonedPayloadsRequest { + int32 limit = 1; +} + +// Server -> client. Carries the batch of tombstoned payloads for the worker to delete. +message GetTombstonedPayloadsResponse { + repeated TombstonedPayload payloads = 1; +} + +// Client -> server. Acknowledges a batch of payloads whose blobs the worker has deleted. +message AckPurgedPayloadsRequest { + repeated PayloadPurgeAck acks = 1; +} + +// Server -> client. Empty response acknowledging the acks were recorded. +message AckPurgedPayloadsResponse { } message GetWorkItemsRequest { diff --git a/test/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs b/test/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs new file mode 100644 index 00000000..126afdf8 --- /dev/null +++ b/test/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs @@ -0,0 +1,100 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using FluentAssertions; +using Microsoft.DurableTask.AzureBlobPayloads; +using Microsoft.DurableTask.Entities.Tests; +using Xunit; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests.AutoPurge; + +public class BlobPurgeJobTests +{ + readonly BlobPurgeJob job = new(new TestLogger()); + + [Fact] + public async Task Create_WhenStopped_ActivatesJobAndStoresBatchSize() + { + // Arrange + TestEntityOperation operation = new( + nameof(BlobPurgeJob.Create), + new TestEntityState(null), + new BlobPurgeJobCreationOptions(250)); + + // Act + await this.job.RunAsync(operation); + + // Assert + BlobPurgeJobState state = Assert.IsType( + operation.State.GetState(typeof(BlobPurgeJobState))); + state.Status.Should().Be(BlobPurgeJobStatus.Active); + state.PurgeBatchSize.Should().Be(250); + state.CreatedAt.Should().NotBeNull(); + state.LastModifiedAt.Should().NotBeNull(); + } + + [Fact] + public async Task Create_WhenAlreadyActive_IsNoOp() + { + // Arrange + BlobPurgeJobState existing = new() + { + Status = BlobPurgeJobStatus.Active, + PurgeBatchSize = 100, + }; + TestEntityOperation operation = new( + nameof(BlobPurgeJob.Create), + new TestEntityState(existing), + new BlobPurgeJobCreationOptions(999)); + + // Act + await this.job.RunAsync(operation); + + // Assert - status stays Active and the original batch size is retained, proving the create no-op'd. + BlobPurgeJobState state = Assert.IsType( + operation.State.GetState(typeof(BlobPurgeJobState))); + state.Status.Should().Be(BlobPurgeJobStatus.Active); + state.PurgeBatchSize.Should().Be(100); + } + + [Fact] + public async Task Create_WithNonPositiveBatchSize_FallsBackToDefault() + { + // Arrange + TestEntityOperation operation = new( + nameof(BlobPurgeJob.Create), + new TestEntityState(null), + new BlobPurgeJobCreationOptions(0)); + + // Act + await this.job.RunAsync(operation); + + // Assert + BlobPurgeJobState state = Assert.IsType( + operation.State.GetState(typeof(BlobPurgeJobState))); + state.PurgeBatchSize.Should().Be(500); + } + + [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/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj b/test/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj new file mode 100644 index 00000000..77078cbb --- /dev/null +++ b/test/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj @@ -0,0 +1,24 @@ + + + + net10.0 + enable + enable + false + true + + Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests + Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests + + + + + + + + + + + + diff --git a/test/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs b/test/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs new file mode 100644 index 00000000..317b1e1e --- /dev/null +++ b/test/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs @@ -0,0 +1,21 @@ +// 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); + } +} diff --git a/test/AzureBlobPayloads.Tests/PayloadStore/BlobPayloadStoreTests.cs b/test/AzureBlobPayloads.Tests/PayloadStore/BlobPayloadStoreTests.cs new file mode 100644 index 00000000..e7fe1448 --- /dev/null +++ b/test/AzureBlobPayloads.Tests/PayloadStore/BlobPayloadStoreTests.cs @@ -0,0 +1,101 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Azure; +using Azure.Storage.Blobs; +using Azure.Storage.Blobs.Models; +using Moq; +using Xunit; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests; + +public class BlobPayloadStoreTests +{ + const string ContainerName = "payloads"; + + static Mock CreateContainer(Mock blob, string expectedBlobName) + { + Mock container = new(); + container.Setup(c => c.Name).Returns(ContainerName); + container.Setup(c => c.GetBlobClient(expectedBlobName)).Returns(blob.Object); + return container; + } + + static Mock CreateBlob(bool existed) + { + Mock blob = new(); + blob + .Setup(b => b.DeleteIfExistsAsync( + It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(Response.FromValue(existed, Mock.Of())); + return blob; + } + + [Fact] + public async Task DeleteAsync_ValidToken_DeletesBackingBlobIncludingSnapshots() + { + // Arrange + Mock blob = CreateBlob(existed: true); + Mock container = CreateContainer(blob, "abc123"); + BlobPayloadStore store = new(container.Object, new LargePayloadStorageOptions()); + + // Act + await store.DeleteAsync($"blob:v1:{ContainerName}:abc123", CancellationToken.None); + + // Assert + container.Verify(c => c.GetBlobClient("abc123"), Times.Once); + blob.Verify( + b => b.DeleteIfExistsAsync(DeleteSnapshotsOption.IncludeSnapshots, null, It.IsAny()), + Times.Once); + } + + [Fact] + public async Task DeleteAsync_MissingBlob_IsIdempotentAndDoesNotThrow() + { + // Arrange + Mock blob = CreateBlob(existed: false); + Mock container = CreateContainer(blob, "missing"); + BlobPayloadStore store = new(container.Object, new LargePayloadStorageOptions()); + + // Act (a missing blob must be a no-op, not an error) + await store.DeleteAsync($"blob:v1:{ContainerName}:missing", CancellationToken.None); + + // Assert + blob.Verify( + b => b.DeleteIfExistsAsync( + It.IsAny(), It.IsAny(), It.IsAny()), + Times.Once); + } + + [Fact] + public async Task DeleteAsync_ContainerMismatch_ThrowsAndDoesNotDelete() + { + // Arrange + Mock blob = CreateBlob(existed: true); + Mock container = CreateContainer(blob, "abc123"); + BlobPayloadStore store = new(container.Object, new LargePayloadStorageOptions()); + + // 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(container.Object, new LargePayloadStorageOptions()); + + // Act & Assert + await Assert.ThrowsAsync(() => store.DeleteAsync(token, CancellationToken.None)); + } +}