From 306d19f4202ec96078b48cd85d4e1c2782915bbf Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 8 Jul 2026 12:50:09 -0700 Subject: [PATCH 1/4] Add opt-in blob payload auto-purge job to AzureBlobPayloads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Large orchestration payloads are externalized to Azure Blob Storage as `blob:v1::` tokens. The DTS backend stores those tokens but cannot delete the backing blobs (it has no storage credentials) — only this SDK can. This adds an opt-in, whole-scheduler singleton durable entity + orchestration job (mirroring src/ExportHistory) that drains payload rows the backend has soft-deleted and deletes their blobs, then acks so the backend can hard-delete the rows. Design: - PayloadStore.DeleteAsync is virtual (default throws NotSupportedException so it is non-breaking for existing external subclasses); BlobPayloadStore overrides it to decode the token and call DeleteIfExistsAsync (idempotent). - BlobPurgeJob (TaskEntity singleton): Create is a no-op when already Active so racing client processes don't disturb the running job; Run starts a fixed-id orchestrator. - BlobPurgeJobOrchestrator (perpetual): fetch a batch of tombstones, delete the blobs with capped parallelism, ack the successful deletions (failed tokens stay tombstoned to retry), idle on a timer when empty, ContinueAsNew periodically. - ExecuteBlobPurgeJobOperationOrchestrator bridges client -> entity. - Two new unary RPCs on TaskHubSidecarService: GetTombstonedPayloads / AckPurgedPayloads (authoritative proto follow-up: microsoft/durabletask-protobuf#76). - LargePayloadStorageOptions gains AutoPurge (opt-in, default false) and PayloadPurgeBatchSize (default 500). - Client-side BlobPurgeJobStarter (IHostedService) ensures the singleton job when AutoPurge is enabled, without blocking host startup. Worker always registers the entity/orchestrators/activities so a client-enabled job has something to run. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- Microsoft.DurableTask.sln | 17 +- src/Client/Core/DurableTaskClient.cs | 24 +++ src/Client/Core/PayloadPurgeAckDto.cs | 13 ++ src/Client/Core/TombstonedPayloadDto.cs | 15 ++ src/Client/Grpc/GrpcDurableTaskClient.cs | 43 ++++++ .../Activities/AckPurgedPayloadsActivity.cs | 36 +++++ .../Activities/DeleteExternalBlobActivity.cs | 42 +++++ .../GetTombstonedPayloadsActivity.cs | 32 ++++ .../AutoPurge/Client/BlobPurgeJobStarter.cs | 135 ++++++++++++++++ .../AutoPurge/Constants/BlobPurgeConstants.cs | 28 ++++ .../AutoPurge/Entity/BlobPurgeJob.cs | 84 ++++++++++ .../AzureBlobPayloads/AutoPurge/Logs.cs | 39 +++++ .../Models/BlobPurgeJobCreationOptions.cs | 12 ++ .../AutoPurge/Models/BlobPurgeJobState.cs | 40 +++++ .../AutoPurge/Models/BlobPurgeJobStatus.cs | 26 ++++ .../BlobPurgeJobOrchestrator.cs | 145 ++++++++++++++++++ ...xecuteBlobPurgeJobOperationOrchestrator.cs | 30 ++++ ...ientBuilderExtensions.AzureBlobPayloads.cs | 38 +++++ ...rkerBuilderExtensions.AzureBlobPayloads.cs | 16 +- .../Options/LargePayloadStorageOptions.cs | 14 ++ .../PayloadStore/BlobPayloadStore.cs | 31 ++++ .../PayloadStore/PayloadStore.cs | 16 ++ src/Grpc/orchestrator_service.proto | 47 ++++++ .../AutoPurge/BlobPurgeJobTests.cs | 100 ++++++++++++ .../AzureBlobPayloads.Tests.csproj | 24 +++ .../LargePayloadStorageOptionsTests.cs | 21 +++ .../PayloadStore/BlobPayloadStoreTests.cs | 101 ++++++++++++ 27 files changed, 1166 insertions(+), 3 deletions(-) create mode 100644 src/Client/Core/PayloadPurgeAckDto.cs create mode 100644 src/Client/Core/TombstonedPayloadDto.cs create mode 100644 src/Extensions/AzureBlobPayloads/AutoPurge/Activities/AckPurgedPayloadsActivity.cs create mode 100644 src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs create mode 100644 src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs create mode 100644 src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs create mode 100644 src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs create mode 100644 src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs create mode 100644 src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs create mode 100644 src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobCreationOptions.cs create mode 100644 src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobState.cs create mode 100644 src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobStatus.cs create mode 100644 src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs create mode 100644 src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/ExecuteBlobPurgeJobOperationOrchestrator.cs create mode 100644 test/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs create mode 100644 test/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj create mode 100644 test/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs create mode 100644 test/AzureBlobPayloads.Tests/PayloadStore/BlobPayloadStoreTests.cs 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)); + } +} From 60f6637bb5983914d0c4de592bf5e30383e693a2 Mon Sep 17 00:00:00 2001 From: wangbill Date: Mon, 13 Jul 2026 18:39:26 -0700 Subject: [PATCH 2/4] Address PR #758 review feedback: naming, self-heal, poison-ack, starter simplification - Drop the `Dto` suffix now that the payload records are first-class public types in `Microsoft.DurableTask.Client` (`TombstonedPayload`, `PayloadPurgeAck`). - Collapse the magic `500` batch-size literal into a single `BlobPurgeConstants.DefaultBatchSize` used everywhere. - Rename `BlobPurgeJobStatus.Stopped` -> `Pending` (still the zero value) and remove the dead `Failed` member (nothing ever set it; the job self-heals). - Make the perpetual orchestrator self-heal: wrap each cycle in try/catch so a transient backend/entity/activity failure logs, backs off, and continues instead of failing the orchestration and killing the eternal loop. - Ack poison tokens: `DeleteExternalBlobActivity` now returns a three-way `BlobDeleteResult` (Deleted/Discarded/Retry). Malformed tokens are discarded and acked so the backend can clear the stuck row instead of re-streaming it forever; transient failures stay tombstoned to retry. - Replace the single-value `BlobPurgeJobCreationOptions` record with a plain `int` on `BlobPurgeJob.Create`. - Guard the client fetch RPC: `GetTombstonedPayloadsAsync` throws `ArgumentOutOfRangeException` unless `0 < limit < 1000`. - Simplify `BlobPurgeJobStarter` to a fixed-instance-id fire-once: drop the entity-active pre-check and schedule the Create bridge once with a fixed instance id, retrying only until the backend is reachable. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/Client/Core/DurableTaskClient.cs | 4 +- ...yloadPurgeAckDto.cs => PayloadPurgeAck.cs} | 2 +- ...onedPayloadDto.cs => TombstonedPayload.cs} | 2 +- src/Client/Grpc/GrpcDurableTaskClient.cs | 16 ++- .../Activities/AckPurgedPayloadsActivity.cs | 4 +- .../Activities/DeleteExternalBlobActivity.cs | 23 ++-- .../GetTombstonedPayloadsActivity.cs | 8 +- .../AutoPurge/Client/BlobPurgeJobStarter.cs | 33 +----- .../AutoPurge/Constants/BlobPurgeConstants.cs | 12 ++ .../AutoPurge/Entity/BlobPurgeJob.cs | 10 +- .../AzureBlobPayloads/AutoPurge/Logs.cs | 6 + .../AutoPurge/Models/BlobDeleteResult.cs | 27 +++++ .../Models/BlobPurgeJobCreationOptions.cs | 12 -- .../AutoPurge/Models/BlobPurgeJobStatus.cs | 11 +- .../BlobPurgeJobOrchestrator.cs | 109 ++++++++++-------- .../Options/LargePayloadStorageOptions.cs | 3 +- .../AutoPurge/BlobPurgeJobTests.cs | 6 +- 17 files changed, 160 insertions(+), 128 deletions(-) rename src/Client/Core/{PayloadPurgeAckDto.cs => PayloadPurgeAck.cs} (87%) rename src/Client/Core/{TombstonedPayloadDto.cs => TombstonedPayload.cs} (87%) create mode 100644 src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobDeleteResult.cs delete mode 100644 src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobCreationOptions.cs diff --git a/src/Client/Core/DurableTaskClient.cs b/src/Client/Core/DurableTaskClient.cs index 208e6441..bc31915c 100644 --- a/src/Client/Core/DurableTaskClient.cs +++ b/src/Client/Core/DurableTaskClient.cs @@ -557,7 +557,7 @@ public virtual Task> ListInstanceIdsAsync( /// 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( + public virtual Task> GetTombstonedPayloadsAsync( int limit, CancellationToken cancellation = default) => throw new NotSupportedException($"{this.GetType()} does not support retrieving tombstoned payloads."); @@ -570,7 +570,7 @@ public virtual Task> GetTombstonedPayloadsAsync( /// 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) + IEnumerable acks, CancellationToken cancellation = default) => throw new NotSupportedException($"{this.GetType()} does not support acknowledging purged payloads."); // TODO: Create task hub diff --git a/src/Client/Core/PayloadPurgeAckDto.cs b/src/Client/Core/PayloadPurgeAck.cs similarity index 87% rename from src/Client/Core/PayloadPurgeAckDto.cs rename to src/Client/Core/PayloadPurgeAck.cs index 01db0f21..6e7788aa 100644 --- a/src/Client/Core/PayloadPurgeAckDto.cs +++ b/src/Client/Core/PayloadPurgeAck.cs @@ -10,4 +10,4 @@ namespace Microsoft.DurableTask.Client; /// 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); +public sealed record PayloadPurgeAck(int PartitionId, long InstanceKey, long PayloadId); diff --git a/src/Client/Core/TombstonedPayloadDto.cs b/src/Client/Core/TombstonedPayload.cs similarity index 87% rename from src/Client/Core/TombstonedPayloadDto.cs rename to src/Client/Core/TombstonedPayload.cs index 69308d47..bb3f6896 100644 --- a/src/Client/Core/TombstonedPayloadDto.cs +++ b/src/Client/Core/TombstonedPayload.cs @@ -12,4 +12,4 @@ namespace Microsoft.DurableTask.Client; /// 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); +public sealed record TombstonedPayload(int PartitionId, long InstanceKey, long PayloadId, string Token); diff --git a/src/Client/Grpc/GrpcDurableTaskClient.cs b/src/Client/Grpc/GrpcDurableTaskClient.cs index 822f7238..8be31fd5 100644 --- a/src/Client/Grpc/GrpcDurableTaskClient.cs +++ b/src/Client/Grpc/GrpcDurableTaskClient.cs @@ -625,17 +625,23 @@ public override async Task> GetOrchestrationHistoryAsync( } /// - public override async Task> GetTombstonedPayloadsAsync( + public override async Task> GetTombstonedPayloadsAsync( int limit, CancellationToken cancellation = default) { + if (limit <= 0 || limit >= 1000) + { + throw new ArgumentOutOfRangeException( + nameof(limit), limit, "Limit must be greater than 0 and less than 1000."); + } + P.GetTombstonedPayloadsResponse response = await this.sidecarClient.GetTombstonedPayloadsAsync( new P.GetTombstonedPayloadsRequest { Limit = limit }, cancellationToken: cancellation); - List result = new(response.Payloads.Count); + List result = new(response.Payloads.Count); foreach (P.TombstonedPayload payload in response.Payloads) { - result.Add(new TombstonedPayloadDto( + result.Add(new TombstonedPayload( payload.PartitionId, payload.InstanceKey, payload.PayloadId, payload.Token)); } @@ -644,12 +650,12 @@ public override async Task> GetTombstonedPayloadsAsyn /// public override async Task AckPurgedPayloadsAsync( - IEnumerable acks, CancellationToken cancellation = default) + IEnumerable acks, CancellationToken cancellation = default) { Check.NotNull(acks); P.AckPurgedPayloadsRequest request = new(); - foreach (PayloadPurgeAckDto ack in acks) + foreach (PayloadPurgeAck ack in acks) { request.Acks.Add(new P.PayloadPurgeAck { diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/AckPurgedPayloadsActivity.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/AckPurgedPayloadsActivity.cs index 637fa5ec..7ef6dbc8 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/AckPurgedPayloadsActivity.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/AckPurgedPayloadsActivity.cs @@ -16,13 +16,13 @@ namespace Microsoft.DurableTask.AzureBlobPayloads; public class AckPurgedPayloadsActivity( DurableTaskClient client, ILogger logger) - : TaskActivity, object?> + : TaskActivity, object?> { readonly DurableTaskClient client = Check.NotNull(client); readonly ILogger logger = Check.NotNull(logger); /// - public override async Task RunAsync(TaskActivityContext context, List input) + public override async Task RunAsync(TaskActivityContext context, List input) { if (input is null || input.Count == 0) { diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs index 0018cb7c..f070f964 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs @@ -7,8 +7,8 @@ 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. +/// re-delivered tokens and concurrent workers are safe. Malformed (poison) tokens are discarded so they get +/// acknowledged instead of retried forever; transient failures leave the payload tombstoned to retry. /// /// The payload store used to delete blobs. /// The logger instance. @@ -16,27 +16,34 @@ namespace Microsoft.DurableTask.AzureBlobPayloads; public class DeleteExternalBlobActivity( PayloadStore store, ILogger logger) - : TaskActivity + : TaskActivity { readonly PayloadStore store = Check.NotNull(store); readonly ILogger logger = Check.NotNull(logger); /// - public override async Task RunAsync(TaskActivityContext context, string input) + public override async Task RunAsync(TaskActivityContext context, string input) { Check.NotNullOrEmpty(input, nameof(input)); try { await this.store.DeleteAsync(input, CancellationToken.None); - return true; + return BlobDeleteResult.Deleted; + } + catch (ArgumentException ex) + { + // The token is malformed or points at a different container; it can never succeed. Discard it so + // the backend clears the row instead of re-streaming the same poison token every cycle. + this.logger.BlobPurgeDeleteDiscarded(ex, input); + return BlobDeleteResult.Discarded; } catch (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. + // Transient failure: leave the payload tombstoned so a later purge cycle can retry it. A single + // bad token must not fail the whole batch. this.logger.BlobPurgeDeleteFailed(ex, input); - return false; + return BlobDeleteResult.Retry; } } } diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs index 6086efe2..2b1bad7d 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs @@ -15,16 +15,16 @@ namespace Microsoft.DurableTask.AzureBlobPayloads; public class GetTombstonedPayloadsActivity( DurableTaskClient client, ILogger logger) - : TaskActivity> + : TaskActivity> { readonly DurableTaskClient client = Check.NotNull(client); readonly ILogger logger = Check.NotNull(logger); /// - public override async Task> RunAsync(TaskActivityContext context, int input) + public override async Task> RunAsync(TaskActivityContext context, int input) { - int limit = input > 0 ? input : 500; - List payloads = + int limit = input > 0 ? input : BlobPurgeConstants.DefaultBatchSize; + 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 index 9652c1e3..17d11ab0 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs @@ -2,7 +2,6 @@ // 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; @@ -51,7 +50,7 @@ public Task StartAsync(CancellationToken cancellationToken) return Task.CompletedTask; } - int batchSize = opts.PayloadPurgeBatchSize > 0 ? opts.PayloadPurgeBatchSize : 500; + int batchSize = opts.PayloadPurgeBatchSize > 0 ? opts.PayloadPurgeBatchSize : BlobPurgeConstants.DefaultBatchSize; // Do not block host startup; ensure the job on a background task with basic retry until the backend // is reachable. @@ -79,19 +78,16 @@ async Task EnsureJobAsync(int batchSize, CancellationToken cancellationToken) { try { - if (await this.IsJobActiveAsync(cancellationToken)) - { - return; - } - + // The singleton is already guaranteed by the entity's fixed key (Create no-ops when the job is + // active) and the orchestrator's fixed instance id, so just schedule the bridge once with a + // fixed instance id. Retry only if the backend is unreachable at startup. BlobPurgeJobOperationRequest request = new( - this.entityId, - nameof(BlobPurgeJob.Create), - new BlobPurgeJobCreationOptions(batchSize)); + this.entityId, nameof(BlobPurgeJob.Create), batchSize); await this.client.ScheduleNewOrchestrationInstanceAsync( new TaskName(nameof(ExecuteBlobPurgeJobOperationOrchestrator)), request, + new StartOrchestrationOptions(BlobPurgeConstants.StarterInstanceId), cancellationToken); this.logger.BlobPurgeJobEnsured(); @@ -115,21 +111,4 @@ await this.client.ScheduleNewOrchestrationInstanceAsync( } } } - - 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 index eafee991..3697f199 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs @@ -14,6 +14,18 @@ static class BlobPurgeConstants /// 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 configured batch size is missing or non-positive. + /// + public const int DefaultBatchSize = 500; + + /// + /// 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}". /// diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs index 8f08b867..c2103fa6 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs @@ -18,11 +18,11 @@ class BlobPurgeJob(ILogger logger) : TaskEntity /// 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) + /// + /// The maximum number of tombstoned payloads to request from the backend per cycle. + /// + public void Create(TaskEntityContext context, int purgeBatchSize) { - Check.NotNull(creationOptions, nameof(creationOptions)); - if (this.State.Status == BlobPurgeJobStatus.Active) { logger.BlobPurgeJobAlreadyRunning(context.Id.Key); @@ -30,7 +30,7 @@ public void Create(TaskEntityContext context, BlobPurgeJobCreationOptions creati } this.State.Status = BlobPurgeJobStatus.Active; - this.State.PurgeBatchSize = creationOptions.PurgeBatchSize > 0 ? creationOptions.PurgeBatchSize : 500; + this.State.PurgeBatchSize = purgeBatchSize > 0 ? purgeBatchSize : BlobPurgeConstants.DefaultBatchSize; this.State.CreatedAt ??= DateTimeOffset.UtcNow; this.State.LastModifiedAt = DateTimeOffset.UtcNow; this.State.LastError = null; diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs index 6e4ebed8..3b7ccc98 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs @@ -36,4 +36,10 @@ static partial class Logs [LoggerMessage(EventId = 818, Level = LogLevel.Warning, Message = "Blob payload auto-purge starter could not ensure the singleton job; retrying.")] public static partial void BlobPurgeStarterRetry(this ILogger logger, Exception exception); + + [LoggerMessage(EventId = 819, Level = LogLevel.Warning, Message = "Discarding poison externalized payload token '{token}'; it can never be deleted, acknowledging it so the backend can clear the row.")] + public static partial void BlobPurgeDeleteDiscarded(this ILogger logger, Exception exception, string token); + + [LoggerMessage(EventId = 820, Level = LogLevel.Warning, Message = "Blob payload auto-purge cycle for job '{jobId}' failed; backing off before retrying so the job keeps running.")] + public static partial void BlobPurgeCycleFailed(this ILogger logger, Exception exception, string? jobId); } diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobDeleteResult.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobDeleteResult.cs new file mode 100644 index 00000000..d3386be5 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobDeleteResult.cs @@ -0,0 +1,27 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.AzureBlobPayloads; + +/// +/// The outcome of attempting to delete a single externalized payload blob during an auto-purge cycle. +/// +public enum BlobDeleteResult +{ + /// + /// The blob was deleted, or was already gone. The payload should be acknowledged so the backend can + /// hard-delete the row. + /// + Deleted, + + /// + /// The token is permanently invalid (poison) and can never be deleted. The payload should still be + /// acknowledged so the backend clears the stuck row instead of re-streaming the same token forever. + /// + Discarded, + + /// + /// A transient failure occurred. The payload is left tombstoned so a later purge cycle can retry it. + /// + Retry, +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobCreationOptions.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobCreationOptions.cs deleted file mode 100644 index e49e423c..00000000 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobCreationOptions.cs +++ /dev/null @@ -1,12 +0,0 @@ -// 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/BlobPurgeJobStatus.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobStatus.cs index 917b8d1b..97c37e66 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobStatus.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Models/BlobPurgeJobStatus.cs @@ -9,18 +9,13 @@ namespace Microsoft.DurableTask.AzureBlobPayloads; 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. + /// The job has not been started yet. This is the default status of a freshly initialized entity, so it is + /// kept as the zero value to avoid a brand-new entity accidentally appearing active. /// - Stopped, + Pending, /// /// 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 index 2b7ba6f2..bf3db1d4 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs @@ -26,8 +26,8 @@ public class BlobPurgeJobOrchestrator : TaskOrchestrator(); string jobId = input.JobEntityId.Key; - int batchSize = input.PurgeBatchSize > 0 ? input.PurgeBatchSize : DefaultPurgeBatchSize; + int batchSize = input.PurgeBatchSize > 0 ? input.PurgeBatchSize : BlobPurgeConstants.DefaultBatchSize; int processedCycles = input.ProcessedCycles; while (true) @@ -54,92 +54,103 @@ public class BlobPurgeJobOrchestrator : TaskOrchestrator( - input.JobEntityId, nameof(BlobPurgeJob.Get), null); - - if (state is null || state.Status != BlobPurgeJobStatus.Active) + try { - logger.BlobPurgeJobOrchestratorStopping(jobId, state?.Status.ToString() ?? "null"); - return null; - } - - List tombstones = await context.CallActivityAsync>( - nameof(GetTombstonedPayloadsActivity), - batchSize, - new TaskOptions(PurgeActivityRetryPolicy)); + // 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; + if (tombstones is null || tombstones.Count == 0) + { + // Nothing to purge right now: block on a timer (push-free idle) then check again. + await context.CreateTimer(IdleDelay, default); + continue; + } + + List acks = await this.DeleteBatchAsync(context, tombstones); + + if (acks.Count > 0) + { + await context.CallActivityAsync( + nameof(AckPurgedPayloadsActivity), + acks, + new TaskOptions(PurgeActivityRetryPolicy)); + + await context.Entities.CallEntityAsync( + input.JobEntityId, nameof(BlobPurgeJob.RecordPurged), (long)acks.Count); + } } - - List deleted = await this.DeleteBatchAsync(context, tombstones); - - if (deleted.Count > 0) + catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException) { - await context.CallActivityAsync( - nameof(AckPurgedPayloadsActivity), - deleted, - new TaskOptions(PurgeActivityRetryPolicy)); - - await context.Entities.CallEntityAsync( - input.JobEntityId, nameof(BlobPurgeJob.RecordPurged), (long)deleted.Count); + // A single bad cycle (transient backend/entity/activity failure) must not kill the perpetual + // loop. Log, back off, then continue so the job self-heals and keeps draining. + logger.BlobPurgeCycleFailed(ex, jobId); + await context.CreateTimer(ErrorBackoff, default); + continue; } } } - async Task> DeleteBatchAsync( - TaskOrchestrationContext context, List tombstones) + async Task> DeleteBatchAsync( + TaskOrchestrationContext context, List tombstones) { - List deleted = new(tombstones.Count); + List acks = new(tombstones.Count); List> tasks = new(); - foreach (TombstonedPayloadDto tombstone in tombstones) + foreach (TombstonedPayload tombstone in tombstones) { tasks.Add(this.DeleteOneAsync(context, tombstone)); if (tasks.Count >= MaxParallelDeletes) { - await DrainAsync(tasks, deleted); + await DrainAsync(tasks, acks); tasks.Clear(); } } if (tasks.Count > 0) { - await DrainAsync(tasks, deleted); + await DrainAsync(tasks, acks); } - return deleted; + return acks; } - static async Task DrainAsync(List> tasks, List deleted) + static async Task DrainAsync(List> tasks, List acks) { 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) + // Acknowledge blobs that were deleted (or already gone) and poison tokens that can never succeed + // so the backend can hard-delete their rows; transient failures stay tombstoned to retry. + if (outcome.ShouldAck) { - deleted.Add(outcome.Ack); + acks.Add(outcome.Ack); } } } - async Task DeleteOneAsync(TaskOrchestrationContext context, TombstonedPayloadDto tombstone) + async Task DeleteOneAsync(TaskOrchestrationContext context, TombstonedPayload tombstone) { - bool deleted = await context.CallActivityAsync( + BlobDeleteResult result = await context.CallActivityAsync( nameof(DeleteExternalBlobActivity), - tombstone.Token, - new TaskOptions(PurgeActivityRetryPolicy)); + tombstone.Token); return new DeleteOutcome( - deleted, - new PayloadPurgeAckDto(tombstone.PartitionId, tombstone.InstanceKey, tombstone.PayloadId)); + result != BlobDeleteResult.Retry, + new PayloadPurgeAck(tombstone.PartitionId, tombstone.InstanceKey, tombstone.PayloadId)); } - readonly record struct DeleteOutcome(bool Deleted, PayloadPurgeAckDto Ack); + readonly record struct DeleteOutcome(bool ShouldAck, PayloadPurgeAck Ack); } diff --git a/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs b/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs index f53c9e72..582e3326 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; @@ -128,5 +129,5 @@ public int ThresholdBytes /// 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; + public int PayloadPurgeBatchSize { get; set; } = BlobPurgeConstants.DefaultBatchSize; } diff --git a/test/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs b/test/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs index 126afdf8..f3a727d6 100644 --- a/test/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs +++ b/test/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs @@ -19,7 +19,7 @@ public async Task Create_WhenStopped_ActivatesJobAndStoresBatchSize() TestEntityOperation operation = new( nameof(BlobPurgeJob.Create), new TestEntityState(null), - new BlobPurgeJobCreationOptions(250)); + 250); // Act await this.job.RunAsync(operation); @@ -45,7 +45,7 @@ public async Task Create_WhenAlreadyActive_IsNoOp() TestEntityOperation operation = new( nameof(BlobPurgeJob.Create), new TestEntityState(existing), - new BlobPurgeJobCreationOptions(999)); + 999); // Act await this.job.RunAsync(operation); @@ -64,7 +64,7 @@ public async Task Create_WithNonPositiveBatchSize_FallsBackToDefault() TestEntityOperation operation = new( nameof(BlobPurgeJob.Create), new TestEntityState(null), - new BlobPurgeJobCreationOptions(0)); + 0); // Act await this.job.RunAsync(operation); From 149c63a9f06ba101bccc701680a7e9fe39d2edfe Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 14 Jul 2026 10:44:23 -0700 Subject: [PATCH 3/4] Refine BlobPurgeJobStarter: pre-check bridge status before rescheduling Avoid needlessly re-running a Completed bridge orchestration on every host restart (fixed id + no dedupe means the backend would purge+replace a terminal instance). Check the existing bridge via GetInstanceAsync and only (re)schedule when it is absent or ended Failed/Terminated, so a failed setup still self-heals. Handle the schedule race with OrchestrationAlreadyExistsException. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../AutoPurge/Client/BlobPurgeJobStarter.cs | 28 +++++++++++++++++-- 1 file changed, 26 insertions(+), 2 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs index 17d11ab0..33a9adec 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs @@ -1,6 +1,7 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. +using DurableTask.Core.Exceptions; using Microsoft.DurableTask.Client; using Microsoft.DurableTask.Entities; using Microsoft.Extensions.Hosting; @@ -79,8 +80,24 @@ async Task EnsureJobAsync(int batchSize, CancellationToken cancellationToken) try { // The singleton is already guaranteed by the entity's fixed key (Create no-ops when the job is - // active) and the orchestrator's fixed instance id, so just schedule the bridge once with a - // fixed instance id. Retry only if the backend is unreachable at startup. + // active) and the orchestrator's fixed instance id. The bridge orchestration's only job is to + // apply the entity's Create once, under a fixed instance id. Before (re)scheduling it, check the + // existing bridge: if it already Completed - or is still alive (Running/Pending/Suspended) - the + // job is set up, so do not reschedule. (Re-running a Completed bridge is wasteful: with a fixed + // id and no dedupe policy the backend would purge and replace the terminal instance on every + // host restart.) Only (re)schedule when the bridge is absent, or ended in a Failed/Terminated + // state that may never have applied Create - which lets a failed setup self-heal. + OrchestrationMetadata? existing = await this.client.GetInstanceAsync( + BlobPurgeConstants.StarterInstanceId, cancellationToken); + + bool needsSchedule = existing is null + or { RuntimeStatus: OrchestrationRuntimeStatus.Failed or OrchestrationRuntimeStatus.Terminated }; + if (!needsSchedule) + { + this.logger.BlobPurgeJobEnsured(); + return; + } + BlobPurgeJobOperationRequest request = new( this.entityId, nameof(BlobPurgeJob.Create), batchSize); @@ -93,6 +110,13 @@ await this.client.ScheduleNewOrchestrationInstanceAsync( this.logger.BlobPurgeJobEnsured(); return; } + catch (OrchestrationAlreadyExistsException) + { + // Race: another client scheduled the bridge between our status check and schedule call. That is + // fine - the singleton is already kicked off; treat it as ensured and stop. + this.logger.BlobPurgeJobEnsured(); + return; + } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { return; From 4d52005c80e35518c7cef3feba2c4f1e5de50b2d Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 14 Jul 2026 11:19:09 -0700 Subject: [PATCH 4/4] Classify RequestFailedException 400 as permanent in DeleteExternalBlobActivity The Azure Storage SDK already retries transient failures internally (connection errors + HTTP 408/429/5xx with backoff), so an escaped exception means those retries were exhausted. Treating every escaped exception as Retry mis-classified permanent service rejections (e.g. Status 400 InvalidResourceName from a malformed decoded blob name) as transient, causing an infinite re-drain of a poison token. Add a RequestFailedException Status 400 -> Discarded branch (ack so the backend clears the row); keep 403/408/429/5xx/timeouts/cancellation as Retry. Document the doc-verified exception model on the activity and add focused tests for the 400 -> Discarded and non-400 -> Retry outcomes. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../Activities/DeleteExternalBlobActivity.cs | 36 +++++++++++- .../DeleteExternalBlobActivityTests.cs | 58 +++++++++++++++++++ 2 files changed, 92 insertions(+), 2 deletions(-) create mode 100644 test/AzureBlobPayloads.Tests/AutoPurge/DeleteExternalBlobActivityTests.cs diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs index f070f964..6452857c 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs @@ -1,15 +1,39 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. +using Azure; using Microsoft.Extensions.Logging; namespace Microsoft.DurableTask.AzureBlobPayloads; /// /// Activity that deletes a single externalized payload blob given its token. Deletion is idempotent, so -/// re-delivered tokens and concurrent workers are safe. Malformed (poison) tokens are discarded so they get -/// acknowledged instead of retried forever; transient failures leave the payload tombstoned to retry. +/// re-delivered tokens and concurrent workers are safe. /// +/// +/// Outcome classification, verified against the Azure.Storage.Blobs / Azure.Core exception model (not +/// assumed): +/// +/// +/// The Azure SDK already retries transient failures internally (connection errors plus HTTP +/// 408/429/500/502/503/504, with exponential backoff), so any exception that escapes +/// means those built-in retries were already exhausted. +/// +/// +/// Permanent failures are discarded (acked so the backend clears the row): an +/// from the store's own token decode / container-mismatch check (thrown client-side before any network call), +/// and a with 400 (for +/// example InvalidUri / InvalidResourceName when the decoded blob name violates Azure naming rules). Retrying +/// either can never succeed. +/// +/// +/// Everything else is treated as transient and leaves the payload tombstoned to retry on a later cycle: +/// throttling / 5xx that outlived the SDK's retries, 403 authorization failures (which need an operator +/// credential fix rather than dropping data), and timeouts / cancellation. A blob is never dropped on an +/// uncertain error, and a single bad token never fails the whole batch. +/// +/// +/// /// The payload store used to delete blobs. /// The logger instance. [DurableTask] @@ -38,6 +62,14 @@ public override async Task RunAsync(TaskActivityContext contex this.logger.BlobPurgeDeleteDiscarded(ex, input); return BlobDeleteResult.Discarded; } + catch (RequestFailedException ex) when (ex.Status == 400) + { + // Service rejected the request as permanently invalid (e.g. InvalidUri / InvalidResourceName - the + // decoded blob name violates Azure naming rules). Retrying can never succeed, so discard it like a + // poison token: ack so the backend clears the row instead of re-streaming it forever. + this.logger.BlobPurgeDeleteDiscarded(ex, input); + return BlobDeleteResult.Discarded; + } catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException) { // Transient failure: leave the payload tombstoned so a later purge cycle can retry it. A single diff --git a/test/AzureBlobPayloads.Tests/AutoPurge/DeleteExternalBlobActivityTests.cs b/test/AzureBlobPayloads.Tests/AutoPurge/DeleteExternalBlobActivityTests.cs new file mode 100644 index 00000000..6ca963cb --- /dev/null +++ b/test/AzureBlobPayloads.Tests/AutoPurge/DeleteExternalBlobActivityTests.cs @@ -0,0 +1,58 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Azure; +using FluentAssertions; +using Microsoft.DurableTask.AzureBlobPayloads; +using Xunit; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests.AutoPurge; + +public class DeleteExternalBlobActivityTests +{ + [Fact] + public async Task RunAsync_WhenDeleteThrowsRequestFailed400_DiscardsPoisonToken() + { + // Arrange - a Status 400 (e.g. InvalidResourceName) is a permanent service rejection. + StubPayloadStore store = new(new RequestFailedException(400, "InvalidResourceName")); + DeleteExternalBlobActivity activity = new(store, new TestLogger()); + + // Act + BlobDeleteResult result = await activity.RunAsync(null!, "blob:v1:payloads:bad name"); + + // Assert - discarded so the backend acks and clears the row instead of re-streaming forever. + result.Should().Be(BlobDeleteResult.Discarded); + } + + [Fact] + public async Task RunAsync_WhenDeleteThrowsRequestFailedNon400_LeavesTombstonedForRetry() + { + // Arrange - a Status 503 that escaped the SDK's internal retries is treated as transient. + StubPayloadStore store = new(new RequestFailedException(503, "ServerBusy")); + DeleteExternalBlobActivity activity = new(store, new TestLogger()); + + // Act + BlobDeleteResult result = await activity.RunAsync(null!, "blob:v1:payloads:abc123"); + + // Assert - left tombstoned so a later purge cycle can retry; a blob is never dropped on doubt. + result.Should().Be(BlobDeleteResult.Retry); + } + + sealed class StubPayloadStore : PayloadStore + { + readonly Exception? deleteError; + + public StubPayloadStore(Exception? deleteError) => this.deleteError = deleteError; + + public override Task DeleteAsync(string token, CancellationToken cancellationToken) => + this.deleteError is null ? Task.CompletedTask : throw this.deleteError; + + public override Task UploadAsync(string payLoad, CancellationToken cancellationToken) => + throw new NotSupportedException(); + + public override Task DownloadAsync(string token, CancellationToken cancellationToken) => + throw new NotSupportedException(); + + public override bool IsKnownPayloadToken(string value) => true; + } +}