From 306d19f4202ec96078b48cd85d4e1c2782915bbf Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 8 Jul 2026 12:50:09 -0700 Subject: [PATCH 01/10] 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 02/10] 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 03/10] 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 04/10] 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; + } +} From 780d7430ae286f533c87c9b01f01f98cdb482a26 Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 14 Jul 2026 13:25:12 -0700 Subject: [PATCH 05/10] Reuse shared PayloadStore and register purge starter conditionally on AutoPurge Address review #B (store duplication) and #C (conditional-DI of the starter): - TryAddSingleton the PayloadStore so we reuse an already-registered shared store instead of creating a second, redundant one. - Register the auto-purge starter only when the configure overload opted into AutoPurge (peek the flag via a probe options instance); UseExternalizedPayloadsCore no longer registers it unconditionally, so the no-arg overload never starts it. - The starter no longer self-checks the flag in StartAsync (registration already gates it); remove the now-unused BlobPurgeDisabled log and refresh its xmldoc. - Add DI-registration tests proving the starter is registered iff AutoPurge is on. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../AutoPurge/Client/BlobPurgeJobStarter.cs | 15 ++--- .../AzureBlobPayloads/AutoPurge/Logs.cs | 3 - ...ientBuilderExtensions.AzureBlobPayloads.cs | 22 +++++-- .../AzureBlobPayloads.Tests.csproj | 1 + .../UseExternalizedPayloadsTests.cs | 62 +++++++++++++++++++ 5 files changed, 86 insertions(+), 17 deletions(-) create mode 100644 test/AzureBlobPayloads.Tests/DependencyInjection/UseExternalizedPayloadsTests.cs diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs index 33a9adec..23176ef8 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs @@ -11,10 +11,11 @@ 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. +/// Client-side hosted service that ensures the singleton blob payload auto-purge job exists. It is registered +/// only when auto-purge is enabled at registration time (see the UseExternalizedPayloads configure overload), +/// so it does not re-check the flag here. 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 { @@ -45,12 +46,6 @@ public BlobPurgeJobStarter( 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 : BlobPurgeConstants.DefaultBatchSize; // Do not block host startup; ensure the job on a background task with basic retry until the backend diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs index 3b7ccc98..b217f53e 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs @@ -28,9 +28,6 @@ static partial class Logs [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); diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs index 2557c6da..78e4d47f 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs @@ -8,6 +8,7 @@ using Microsoft.DurableTask.Converters; using Microsoft.DurableTask.Worker.Grpc.Internal; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -33,13 +34,28 @@ public static IDurableTaskClientBuilder UseExternalizedPayloads( Check.NotNull(configure); builder.Services.Configure(builder.Name, configure); - builder.Services.AddSingleton(sp => + + // Reuse the shared payload store when one is already registered (e.g. via AddExternalizedPayloadStore or + // the worker builder in the same process); only register our own as a fallback so we never create a + // second, redundant PayloadStore. + builder.Services.TryAddSingleton(sp => { LargePayloadStorageOptions opts = sp.GetRequiredService>().Get(builder.Name); return new BlobPayloadStore(opts); }); - return UseExternalizedPayloadsCore(builder); + UseExternalizedPayloadsCore(builder); + + // Conditional DI: register the auto-purge starter only when the caller opted into auto-purge. Peek the + // flag now by running the configure delegate against a probe (options configurators are pure setters). + LargePayloadStorageOptions probe = new(); + configure(probe); + if (probe.AutoPurge) + { + RegisterBlobPurgeJobStarter(builder); + } + + return builder; } /// @@ -82,8 +98,6 @@ static IDurableTaskClientBuilder UseExternalizedPayloadsCore(IDurableTaskClientB } }); - RegisterBlobPurgeJobStarter(builder); - return builder; } diff --git a/test/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj b/test/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj index 77078cbb..b650cb59 100644 --- a/test/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj +++ b/test/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj @@ -14,6 +14,7 @@ + diff --git a/test/AzureBlobPayloads.Tests/DependencyInjection/UseExternalizedPayloadsTests.cs b/test/AzureBlobPayloads.Tests/DependencyInjection/UseExternalizedPayloadsTests.cs new file mode 100644 index 00000000..b3343078 --- /dev/null +++ b/test/AzureBlobPayloads.Tests/DependencyInjection/UseExternalizedPayloadsTests.cs @@ -0,0 +1,62 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using FluentAssertions; +using Microsoft.DurableTask.Client; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Moq; +using Xunit; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests.DependencyInjection; + +public class UseExternalizedPayloadsTests +{ + [Fact] + public void UseExternalizedPayloads_WithAutoPurgeEnabled_RegistersHostedPurgeStarter() + { + // Arrange + ServiceCollection services = new(); + Mock builder = new(); + builder.Setup(b => b.Services).Returns(services); + builder.Setup(b => b.Name).Returns(string.Empty); + + // Act + builder.Object.UseExternalizedPayloads(options => options.AutoPurge = true); + + // Assert - the purge-job starter is the only IHostedService this path registers. + services.Should().ContainSingle(d => d.ServiceType == typeof(IHostedService)); + } + + [Fact] + public void UseExternalizedPayloads_WithAutoPurgeDisabled_DoesNotRegisterHostedPurgeStarter() + { + // Arrange + ServiceCollection services = new(); + Mock builder = new(); + builder.Setup(b => b.Services).Returns(services); + builder.Setup(b => b.Name).Returns(string.Empty); + + // Act - auto-purge left at its default (false). + builder.Object.UseExternalizedPayloads(options => { }); + + // Assert + services.Should().NotContain(d => d.ServiceType == typeof(IHostedService)); + } + + [Fact] + public void UseExternalizedPayloads_NoArgOverload_DoesNotRegisterHostedPurgeStarter() + { + // Arrange + ServiceCollection services = new(); + Mock builder = new(); + builder.Setup(b => b.Services).Returns(services); + builder.Setup(b => b.Name).Returns(string.Empty); + + // Act - the shared-store overload never enables auto-purge. + builder.Object.UseExternalizedPayloads(); + + // Assert + services.Should().NotContain(d => d.ServiceType == typeof(IHostedService)); + } +} From 3a2215cd9796268ef56064f4603f1a3964fdc17f Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 14 Jul 2026 14:07:32 -0700 Subject: [PATCH 06/10] Register fallback PayloadStore in shared Core for both client and worker The store's consumer is the interceptor wired up in UseExternalizedPayloadsCore, so Core is the single fallback registration site. Move the TryAddSingleton out of each configure overload and into Core, symmetrically for the client and worker extensions. TryAdd keeps reusing a shared store (AddExternalizedPayloadStore or the sibling builder) and never creates a redundant one. The worker overload switches from AddSingleton to the shared TryAddSingleton in Core as part of the move. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- ...lientBuilderExtensions.AzureBlobPayloads.cs | 18 +++++++++--------- ...orkerBuilderExtensions.AzureBlobPayloads.cs | 15 ++++++++++----- 2 files changed, 19 insertions(+), 14 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs index 78e4d47f..0c485208 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs @@ -35,15 +35,6 @@ public static IDurableTaskClientBuilder UseExternalizedPayloads( builder.Services.Configure(builder.Name, configure); - // Reuse the shared payload store when one is already registered (e.g. via AddExternalizedPayloadStore or - // the worker builder in the same process); only register our own as a fallback so we never create a - // second, redundant PayloadStore. - builder.Services.TryAddSingleton(sp => - { - LargePayloadStorageOptions opts = sp.GetRequiredService>().Get(builder.Name); - return new BlobPayloadStore(opts); - }); - UseExternalizedPayloadsCore(builder); // Conditional DI: register the auto-purge starter only when the caller opted into auto-purge. Peek the @@ -73,6 +64,15 @@ public static IDurableTaskClientBuilder UseExternalizedPayloads( static IDurableTaskClientBuilder UseExternalizedPayloadsCore(IDurableTaskClientBuilder builder) { + // Reuse the shared payload store when one is already registered (e.g. via AddExternalizedPayloadStore or + // the worker builder in the same process); only register our own as a fallback so we never create a + // second, redundant PayloadStore. + builder.Services.TryAddSingleton(sp => + { + LargePayloadStorageOptions opts = sp.GetRequiredService>().Get(builder.Name); + return new BlobPayloadStore(opts); + }); + // Wrap the gRPC CallInvoker with our interceptor when using the gRPC client builder.Services .AddOptions(builder.Name) diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs index 42ac0f0a..c63db137 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs @@ -6,6 +6,7 @@ using Microsoft.DurableTask.Worker; using Microsoft.DurableTask.Worker.Grpc; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Options; using P = Microsoft.DurableTask.Protobuf; @@ -30,11 +31,6 @@ public static IDurableTaskWorkerBuilder UseExternalizedPayloads( Check.NotNull(configure); builder.Services.Configure(builder.Name, configure); - builder.Services.AddSingleton(sp => - { - LargePayloadStorageOptions opts = sp.GetRequiredService>().Get(builder.Name); - return new BlobPayloadStore(opts); - }); return UseExternalizedPayloadsCore(builder); } @@ -54,6 +50,15 @@ public static IDurableTaskWorkerBuilder UseExternalizedPayloads( static IDurableTaskWorkerBuilder UseExternalizedPayloadsCore(IDurableTaskWorkerBuilder builder) { + // Reuse the shared payload store when one is already registered (e.g. via AddExternalizedPayloadStore or + // the client builder in the same process); only register our own as a fallback so we never create a + // second, redundant PayloadStore. + builder.Services.TryAddSingleton(sp => + { + LargePayloadStorageOptions opts = sp.GetRequiredService>().Get(builder.Name); + return new BlobPayloadStore(opts); + }); + // Wrap the gRPC CallInvoker with our interceptor when using the gRPC worker builder.Services .AddOptions(builder.Name) From 47651dc0e93d17a67dc145f14d028df019782590 Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 14 Jul 2026 14:39:58 -0700 Subject: [PATCH 07/10] Stop self-registering PayloadStore on the client; consume the shared store Restore the original design: the client never registers a PayloadStore, it only consumes a shared/external one (via AddExternalizedPayloadStore or an in-process worker). Only the worker self-registers a fallback store. Remove the client Core's TryAddSingleton block (and its now-unused Microsoft.Extensions.DependencyInjection.Extensions using); the client Core PostConfigure still resolves PayloadStore from the shared/worker registration. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- ...bleTaskClientBuilderExtensions.AzureBlobPayloads.cs | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs index 0c485208..f811d6da 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs @@ -8,7 +8,6 @@ using Microsoft.DurableTask.Converters; using Microsoft.DurableTask.Worker.Grpc.Internal; using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -64,15 +63,6 @@ public static IDurableTaskClientBuilder UseExternalizedPayloads( static IDurableTaskClientBuilder UseExternalizedPayloadsCore(IDurableTaskClientBuilder builder) { - // Reuse the shared payload store when one is already registered (e.g. via AddExternalizedPayloadStore or - // the worker builder in the same process); only register our own as a fallback so we never create a - // second, redundant PayloadStore. - builder.Services.TryAddSingleton(sp => - { - LargePayloadStorageOptions opts = sp.GetRequiredService>().Get(builder.Name); - return new BlobPayloadStore(opts); - }); - // Wrap the gRPC CallInvoker with our interceptor when using the gRPC client builder.Services .AddOptions(builder.Name) From 7397fa64e71b88150087b551580944dce0e8cc2f Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 15 Jul 2026 10:02:11 -0700 Subject: [PATCH 08/10] Validate PayloadPurgeBatchSize once at specification (fail fast on out-of-range); drop redundant downstream coercions Validate the batch size at its single point of specification - LargePayloadStorageOptions.PayloadPurgeBatchSize - and fail fast when out of range, mirroring the existing ThresholdBytes setter. Valid range is 1..999 inclusive, matching the gRPC GetTombstonedPayloadsAsync contract (which rejects limits >= 1000); add BlobPurgeConstants.MaxBatchSize = 999 to express the upper bound. Now that the value is guaranteed valid at specification, remove the redundant '> 0 ? x : DefaultBatchSize' coercions from every downstream consumer (starter, entity, orchestrator, activity); they simply use the value directly. Tests: add setter validation tests (throws for 0, -1, 1000, 1001; accepts 1, 500, 999) and repurpose the entity's non-positive test to assert the entity now stores the batch size verbatim (no coercion). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../GetTombstonedPayloadsActivity.cs | 2 +- .../AutoPurge/Client/BlobPurgeJobStarter.cs | 2 +- .../AutoPurge/Constants/BlobPurgeConstants.cs | 8 ++++- .../AutoPurge/Entity/BlobPurgeJob.cs | 2 +- .../BlobPurgeJobOrchestrator.cs | 2 +- .../Options/LargePayloadStorageOptions.cs | 20 +++++++++-- .../AutoPurge/BlobPurgeJobTests.cs | 8 +++-- .../LargePayloadStorageOptionsTests.cs | 33 +++++++++++++++++++ 8 files changed, 67 insertions(+), 10 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs index 2b1bad7d..838b293a 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Activities/GetTombstonedPayloadsActivity.cs @@ -23,7 +23,7 @@ public class GetTombstonedPayloadsActivity( /// public override async Task> RunAsync(TaskActivityContext context, int input) { - int limit = input > 0 ? input : BlobPurgeConstants.DefaultBatchSize; + int limit = input; List payloads = await this.client.GetTombstonedPayloadsAsync(limit, CancellationToken.None); this.logger.BlobPurgeFetchedTombstones(payloads.Count); diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs index 23176ef8..bc3e18a5 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs @@ -46,7 +46,7 @@ public BlobPurgeJobStarter( public Task StartAsync(CancellationToken cancellationToken) { LargePayloadStorageOptions opts = this.options.Get(this.builderName); - int batchSize = opts.PayloadPurgeBatchSize > 0 ? opts.PayloadPurgeBatchSize : BlobPurgeConstants.DefaultBatchSize; + int batchSize = opts.PayloadPurgeBatchSize; // Do not block host startup; ensure the job on a background task with basic retry until the backend // is reachable. diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs index 3697f199..48174f52 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs @@ -16,10 +16,16 @@ static class BlobPurgeConstants /// /// 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. + /// used whenever a batch size is not explicitly configured. /// public const int DefaultBatchSize = 500; + /// + /// The maximum batch size the auto-purge job may request per cycle. Mirrors the gRPC + /// GetTombstonedPayloadsAsync contract, which rejects limits >= 1000. + /// + public const int MaxBatchSize = 999; + /// /// 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. diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs index c2103fa6..89391890 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Entity/BlobPurgeJob.cs @@ -30,7 +30,7 @@ public void Create(TaskEntityContext context, int purgeBatchSize) } this.State.Status = BlobPurgeJobStatus.Active; - this.State.PurgeBatchSize = purgeBatchSize > 0 ? purgeBatchSize : BlobPurgeConstants.DefaultBatchSize; + this.State.PurgeBatchSize = purgeBatchSize; this.State.CreatedAt ??= DateTimeOffset.UtcNow; this.State.LastModifiedAt = DateTimeOffset.UtcNow; this.State.LastError = null; diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs index bf3db1d4..8ba4f707 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs @@ -42,7 +42,7 @@ public class BlobPurgeJobOrchestrator : TaskOrchestrator(); string jobId = input.JobEntityId.Key; - int batchSize = input.PurgeBatchSize > 0 ? input.PurgeBatchSize : BlobPurgeConstants.DefaultBatchSize; + int batchSize = input.PurgeBatchSize; int processedCycles = input.ProcessedCycles; while (true) diff --git a/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs b/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs index 582e3326..cbd63835 100644 --- a/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs +++ b/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs @@ -28,6 +28,7 @@ namespace Microsoft.DurableTask; public sealed class LargePayloadStorageOptions { int thresholdBytes = 256 * 1024; + int payloadPurgeBatchSize = BlobPurgeConstants.DefaultBatchSize; /// /// Initializes a new instance of the class. @@ -127,7 +128,22 @@ 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. + /// per cycle. Must be between 1 and 999 (inclusive); values outside this range throw + /// . Defaults to 500. /// - public int PayloadPurgeBatchSize { get; set; } = BlobPurgeConstants.DefaultBatchSize; + public int PayloadPurgeBatchSize + { + get => this.payloadPurgeBatchSize; + set + { + if (value < 1 || value > BlobPurgeConstants.MaxBatchSize) + { + throw new ArgumentOutOfRangeException( + nameof(this.PayloadPurgeBatchSize), value, + $"PayloadPurgeBatchSize must be between 1 and {BlobPurgeConstants.MaxBatchSize} (inclusive)."); + } + + this.payloadPurgeBatchSize = value; + } + } } diff --git a/test/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs b/test/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs index f3a727d6..2117299c 100644 --- a/test/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs +++ b/test/AzureBlobPayloads.Tests/AutoPurge/BlobPurgeJobTests.cs @@ -58,9 +58,11 @@ public async Task Create_WhenAlreadyActive_IsNoOp() } [Fact] - public async Task Create_WithNonPositiveBatchSize_FallsBackToDefault() + public async Task Create_StoresBatchSizeVerbatim_WithoutCoercion() { - // Arrange + // Arrange - the batch size is validated once at specification (LargePayloadStorageOptions), so the + // entity trusts its input and performs no coercion of its own. A zero here is stored as-is, proving + // the previous non-positive-to-default fallback was removed. TestEntityOperation operation = new( nameof(BlobPurgeJob.Create), new TestEntityState(null), @@ -72,7 +74,7 @@ public async Task Create_WithNonPositiveBatchSize_FallsBackToDefault() // Assert BlobPurgeJobState state = Assert.IsType( operation.State.GetState(typeof(BlobPurgeJobState))); - state.PurgeBatchSize.Should().Be(500); + state.PurgeBatchSize.Should().Be(0); } [Fact] diff --git a/test/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs b/test/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs index 317b1e1e..6a4a95ed 100644 --- a/test/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs +++ b/test/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs @@ -18,4 +18,37 @@ public void Defaults_AutoPurgeDisabled_AndBatchSize500() options.AutoPurge.Should().BeFalse(); options.PayloadPurgeBatchSize.Should().Be(500); } + + [Theory] + [InlineData(0)] + [InlineData(-1)] + [InlineData(1000)] + [InlineData(1001)] + public void PayloadPurgeBatchSize_OutOfRange_Throws(int value) + { + // Arrange + LargePayloadStorageOptions options = new(); + + // Act + Action act = () => options.PayloadPurgeBatchSize = value; + + // Assert + act.Should().Throw(); + } + + [Theory] + [InlineData(1)] + [InlineData(500)] + [InlineData(999)] + public void PayloadPurgeBatchSize_InRange_IsAccepted(int value) + { + // Arrange + LargePayloadStorageOptions options = new(); + + // Act + options.PayloadPurgeBatchSize = value; + + // Assert + options.PayloadPurgeBatchSize.Should().Be(value); + } } From 4afeb8a93aa05cb46a3098c4f0d54656c583c938 Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 15 Jul 2026 10:03:51 -0700 Subject: [PATCH 09/10] Translate gRPC Cancelled to OperationCanceledException in GetTombstonedPayloadsAsync/AckPurgedPayloadsAsync Both new methods previously let RpcException(StatusCode.Cancelled) escape, unlike every other method in this client. Wrap each method's gRPC call in a try/catch that translates Cancelled to OperationCanceledException, mirroring the existing pattern used by GetOrchestrationHistoryAsync and the other client methods. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/Client/Grpc/GrpcDurableTaskClient.cs | 25 ++++++++++++++++++++---- 1 file changed, 21 insertions(+), 4 deletions(-) diff --git a/src/Client/Grpc/GrpcDurableTaskClient.cs b/src/Client/Grpc/GrpcDurableTaskClient.cs index 8be31fd5..ca9b51ae 100644 --- a/src/Client/Grpc/GrpcDurableTaskClient.cs +++ b/src/Client/Grpc/GrpcDurableTaskClient.cs @@ -634,9 +634,18 @@ public override async Task> GetTombstonedPayloadsAsync( 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); + P.GetTombstonedPayloadsResponse response; + try + { + response = await this.sidecarClient.GetTombstonedPayloadsAsync( + new P.GetTombstonedPayloadsRequest { Limit = limit }, + cancellationToken: cancellation); + } + catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled) + { + throw new OperationCanceledException( + $"The {nameof(this.GetTombstonedPayloadsAsync)} operation was canceled.", e, cancellation); + } List result = new(response.Payloads.Count); foreach (P.TombstonedPayload payload in response.Payloads) @@ -670,7 +679,15 @@ public override async Task AckPurgedPayloadsAsync( return; } - await this.sidecarClient.AckPurgedPayloadsAsync(request, cancellationToken: cancellation); + try + { + await this.sidecarClient.AckPurgedPayloadsAsync(request, cancellationToken: cancellation); + } + catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled) + { + throw new OperationCanceledException( + $"The {nameof(this.AckPurgedPayloadsAsync)} operation was canceled.", e, cancellation); + } } static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, ILogger logger, out CallInvoker callInvoker) From e74f6330ab6de3fe6accdfab1892a7e6752e5a9a Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 15 Jul 2026 11:05:17 -0700 Subject: [PATCH 10/10] Raise auto-purge MaxBatchSize to 1000 (inclusive); relax gRPC GetTombstonedPayloads limit check to allow 1000 1000 is a valid, accepted batch size end-to-end: the proto documents no numeric range and the backend only clamps via min(limit, configuredCap) without rejecting >= 1000, so the whole range is contained in this SDK. Raise BlobPurgeConstants.MaxBatchSize from 999 to 1000 (the options setter already validates against it) and relax the gRPC GetTombstonedPayloadsAsync guard from 'limit >= 1000' to 'limit > 1000'. Update the two PayloadPurgeBatchSize theories so 1000 is asserted accepted and 1001 still throws. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/Client/Grpc/GrpcDurableTaskClient.cs | 4 ++-- .../AutoPurge/Constants/BlobPurgeConstants.cs | 4 ++-- .../AzureBlobPayloads/Options/LargePayloadStorageOptions.cs | 2 +- .../Options/LargePayloadStorageOptionsTests.cs | 2 +- 4 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/Client/Grpc/GrpcDurableTaskClient.cs b/src/Client/Grpc/GrpcDurableTaskClient.cs index ca9b51ae..6e79b6b2 100644 --- a/src/Client/Grpc/GrpcDurableTaskClient.cs +++ b/src/Client/Grpc/GrpcDurableTaskClient.cs @@ -628,10 +628,10 @@ public override async Task> GetOrchestrationHistoryAsync( public override async Task> GetTombstonedPayloadsAsync( int limit, CancellationToken cancellation = default) { - if (limit <= 0 || limit >= 1000) + if (limit <= 0 || limit > 1000) { throw new ArgumentOutOfRangeException( - nameof(limit), limit, "Limit must be greater than 0 and less than 1000."); + nameof(limit), limit, "Limit must be greater than 0 and less than or equal to 1000."); } P.GetTombstonedPayloadsResponse response; diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs index 48174f52..ca4cce32 100644 --- a/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Constants/BlobPurgeConstants.cs @@ -22,9 +22,9 @@ static class BlobPurgeConstants /// /// The maximum batch size the auto-purge job may request per cycle. Mirrors the gRPC - /// GetTombstonedPayloadsAsync contract, which rejects limits >= 1000. + /// GetTombstonedPayloadsAsync contract, which rejects limits greater than 1000. /// - public const int MaxBatchSize = 999; + public const int MaxBatchSize = 1000; /// /// The fixed instance ID of the client-to-entity bridge orchestration the starter schedules to ensure the diff --git a/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs b/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs index cbd63835..0edaee90 100644 --- a/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs +++ b/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs @@ -128,7 +128,7 @@ public int ThresholdBytes /// /// Gets or sets the maximum number of tombstoned payloads the auto-purge job requests from the backend - /// per cycle. Must be between 1 and 999 (inclusive); values outside this range throw + /// per cycle. Must be between 1 and 1000 (inclusive); values outside this range throw /// . Defaults to 500. /// public int PayloadPurgeBatchSize diff --git a/test/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs b/test/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs index 6a4a95ed..d608a936 100644 --- a/test/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs +++ b/test/AzureBlobPayloads.Tests/Options/LargePayloadStorageOptionsTests.cs @@ -22,7 +22,6 @@ public void Defaults_AutoPurgeDisabled_AndBatchSize500() [Theory] [InlineData(0)] [InlineData(-1)] - [InlineData(1000)] [InlineData(1001)] public void PayloadPurgeBatchSize_OutOfRange_Throws(int value) { @@ -40,6 +39,7 @@ public void PayloadPurgeBatchSize_OutOfRange_Throws(int value) [InlineData(1)] [InlineData(500)] [InlineData(999)] + [InlineData(1000)] public void PayloadPurgeBatchSize_InRange_IsAccepted(int value) { // Arrange