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/Extensions/AzureBlobPayloads/AutoPurge/BlobPayloadAutoPurgeService.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/BlobPayloadAutoPurgeService.cs new file mode 100644 index 00000000..27a107f6 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/BlobPayloadAutoPurgeService.cs @@ -0,0 +1,233 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Grpc.Core; +using Microsoft.DurableTask.Worker.Grpc; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask; + +/// +/// Background service that drains blob-externalized payload tombstones streamed by the Durable Task +/// backend and deletes the corresponding blobs from customer storage. The backend cannot delete these +/// blobs itself because it has no storage credentials; it soft-deletes the payload row, pushes the token +/// to this worker, and hard-deletes the row only once the worker acknowledges the blob has been removed. +/// +/// +/// +/// On the open-source worker this service is registered automatically by UseExternalizedPayloads +/// and reuses that worker's gRPC connection. It can also be constructed directly from a +/// (or gRPC channel) and a so hosts that enable large +/// payloads through a different code path (for example the Durable Task Scheduler "Azure Managed" worker +/// SDK) can reuse it without going through UseExternalizedPayloads. Because it derives from +/// such hosts can start and stop it directly via +/// / . +/// +/// +/// It opens a single bidirectional gRPC stream and blocks awaiting server pushes when idle (a push model, +/// not a polling loop) for the worker process lifetime. Deletes are idempotent, so running multiple worker +/// replicas concurrently is safe. +/// +/// +public sealed class BlobPayloadAutoPurgeService : BackgroundService +{ + static readonly TimeSpan ReconnectBackoffBase = TimeSpan.FromSeconds(1); + static readonly TimeSpan ReconnectBackoffCap = TimeSpan.FromSeconds(30); + + readonly PayloadStore store; + readonly ILogger logger; + readonly Random jitter = new(); + + // Exactly one connection strategy is used. When a CallInvoker is supplied directly (cross-repo reuse) it + // is used as-is; otherwise the OSS DI path resolves it lazily from the named worker options at start, + // which keeps host startup resilient if the worker connection is misconfigured. + readonly CallInvoker? callInvoker; + readonly IOptionsMonitor? workerOptions; + readonly string? builderName; + + /// + /// Initializes a new instance of the class that drains the + /// purge stream over the supplied gRPC call invoker. Use this constructor to reuse the service from a + /// host that does not enable large payloads through UseExternalizedPayloads. + /// + /// The gRPC call invoker connected to the Durable Task backend. + /// The payload store used to delete blobs. + /// The logger, or null to disable logging. + public BlobPayloadAutoPurgeService( + CallInvoker callInvoker, + PayloadStore store, + ILogger? logger = null) + : this(store, logger) + { + this.callInvoker = Check.NotNull(callInvoker); + } + + /// + /// Initializes a new instance of the class that drains the + /// purge stream over the supplied gRPC channel. Use this constructor to reuse the service from a host + /// that does not enable large payloads through UseExternalizedPayloads. + /// + /// The gRPC channel connected to the Durable Task backend. + /// The payload store used to delete blobs. + /// The logger, or null to disable logging. + public BlobPayloadAutoPurgeService( + ChannelBase channel, + PayloadStore store, + ILogger? logger = null) + : this(Check.NotNull(channel).CreateCallInvoker(), store, logger) + { + } + + /// + /// Initializes a new instance of the class that resolves its + /// gRPC connection from the named worker options. Used by the UseExternalizedPayloads DI path. + /// + /// The payload store used to delete blobs. + /// Monitor for the named gRPC worker options that carry the DTS connection. + /// The Durable Task worker builder name whose options this service uses. + /// The logger. + internal BlobPayloadAutoPurgeService( + PayloadStore store, + IOptionsMonitor workerOptions, + string builderName, + ILogger logger) + : this(store, logger) + { + this.workerOptions = Check.NotNull(workerOptions); + this.builderName = Check.NotNull(builderName); + } + + BlobPayloadAutoPurgeService(PayloadStore store, ILogger? logger) + { + this.store = Check.NotNull(store); + this.logger = logger ?? NullLogger.Instance; + } + + /// + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + CallInvoker? callInvoker = this.ResolveCallInvoker(); + if (callInvoker is null) + { + // ResolveCallInvoker already logged why the purge stream cannot run. + return; + } + + P.TaskHubSidecarService.TaskHubSidecarServiceClient client = new(callInvoker); + TimeSpan backoff = ReconnectBackoffBase; + + while (!stoppingToken.IsCancellationRequested) + { + try + { + await this.DrainPurgeStreamAsync(client, stoppingToken); + + // A clean server-side completion just means "nothing more to purge right now"; reconnect promptly. + backoff = ReconnectBackoffBase; + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException) + { + Logs.AutoPurgeStreamError(this.logger, ex); + await this.DelayWithJitterAsync(backoff, stoppingToken); + backoff = NextBackoff(backoff); + } + } + } + + static TimeSpan NextBackoff(TimeSpan current) + { + if (current >= ReconnectBackoffCap) + { + return ReconnectBackoffCap; + } + + long doubledTicks = current.Ticks * 2; + return doubledTicks >= ReconnectBackoffCap.Ticks ? ReconnectBackoffCap : TimeSpan.FromTicks(doubledTicks); + } + + CallInvoker? ResolveCallInvoker() + { + // Cross-repo reuse path: the caller supplied a ready-to-use connection. + if (this.callInvoker is not null) + { + return this.callInvoker; + } + + // OSS DI path: reuse the same connection the worker uses to reach DTS. After UseExternalizedPayloads + // runs, the named options are guaranteed to expose a CallInvoker (a Channel, if supplied, is folded + // into one during PostConfigure). + CallInvoker? resolved; + try + { + resolved = this.workerOptions?.Get(this.builderName!).CallInvoker; + } + catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException) + { + Logs.AutoPurgeConfigurationError(this.logger, ex); + return null; + } + + if (resolved is null) + { + Logs.AutoPurgeNoCallInvoker(this.logger); + } + + return resolved; + } + + async Task DrainPurgeStreamAsync( + P.TaskHubSidecarService.TaskHubSidecarServiceClient client, CancellationToken stoppingToken) + { + using AsyncDuplexStreamingCall call = + client.PurgeExternalPayloads(cancellationToken: stoppingToken); + + Logs.AutoPurgeStreamOpened(this.logger); + + while (await call.ResponseStream.MoveNext(stoppingToken)) + { + P.TombstonedPayload tombstone = call.ResponseStream.Current; + try + { + await this.store.DeleteAsync(tombstone.Token, stoppingToken); + } + catch (Exception ex) when ( + ex is not OperationCanceledException and not OutOfMemoryException and not StackOverflowException) + { + // Leave the payload soft-deleted so the backend can re-push it later; skip the ack. + Logs.AutoPurgeDeleteFailed(this.logger, ex, tombstone.Token); + continue; + } + + // The ack write inherits cancellation from the duplex call (opened with stoppingToken); the + // WriteAsync(message, CancellationToken) overload is unavailable on netstandard2.0. +#pragma warning disable CA2016 // Forward the CancellationToken parameter to methods that take one + await call.RequestStream.WriteAsync(new P.PayloadPurgeAck + { + PartitionId = tombstone.PartitionId, + InstanceKey = tombstone.InstanceKey, + PayloadId = tombstone.PayloadId, + }); +#pragma warning restore CA2016 + } + + await call.RequestStream.CompleteAsync(); + } + + async Task DelayWithJitterAsync(TimeSpan maxDelay, CancellationToken cancellationToken) + { + // Full jitter in [0, maxDelay) spreads reconnect attempts across replicas. + TimeSpan delay = TimeSpan.FromTicks((long)(maxDelay.Ticks * this.jitter.NextDouble())); + if (delay > TimeSpan.Zero) + { + await Task.Delay(delay, cancellationToken); + } + } +} diff --git a/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs b/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs new file mode 100644 index 00000000..0925f035 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AutoPurge/Logs.cs @@ -0,0 +1,42 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.Extensions.Logging; + +namespace Microsoft.DurableTask; + +/// +/// Log messages for the Azure Blob externalized-payload auto-purge background service. +/// +static partial class Logs +{ + [LoggerMessage( + EventId = 800, + Level = LogLevel.Debug, + Message = "Blob payload auto-purge stream opened; awaiting tombstoned payloads from the backend.")] + public static partial void AutoPurgeStreamOpened(ILogger logger); + + [LoggerMessage( + EventId = 801, + Level = LogLevel.Warning, + Message = "Blob payload auto-purge stream failed; reconnecting after backoff.")] + public static partial void AutoPurgeStreamError(ILogger logger, Exception exception); + + [LoggerMessage( + EventId = 802, + Level = LogLevel.Warning, + Message = "Failed to delete externalized payload blob for token '{Token}'; leaving it for the backend to re-push.")] + public static partial void AutoPurgeDeleteFailed(ILogger logger, Exception exception, string token); + + [LoggerMessage( + EventId = 803, + Level = LogLevel.Warning, + Message = "Blob payload auto-purge could not resolve a gRPC call invoker; the purge stream will not run.")] + public static partial void AutoPurgeNoCallInvoker(ILogger logger); + + [LoggerMessage( + EventId = 804, + Level = LogLevel.Error, + Message = "Blob payload auto-purge failed to read the worker gRPC options; the purge stream will not run.")] + public static partial void AutoPurgeConfigurationError(ILogger logger, Exception exception); +} diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs index b690d288..577f24b4 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs @@ -7,6 +7,8 @@ using Microsoft.DurableTask.Worker; using Microsoft.DurableTask.Worker.Grpc; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using P = Microsoft.DurableTask.Protobuf; @@ -82,6 +84,20 @@ static IDurableTaskWorkerBuilder UseExternalizedPayloadsCore(IDurableTaskWorkerB opt.Capabilities.Add(P.WorkerCapability.LargePayloads); }); + // Drain blob-externalized payload tombstones the backend streams for deletion. Gated on this enable + // path: if UseExternalizedPayloads is never called, the service is never registered and never runs. + builder.Services.AddSingleton( + sp => CreateAutoPurgeService(sp, builder.Name)); + return builder; } + + static BlobPayloadAutoPurgeService CreateAutoPurgeService(IServiceProvider services, string builderName) + { + return new BlobPayloadAutoPurgeService( + services.GetRequiredService(), + services.GetRequiredService>(), + builderName, + services.GetRequiredService>()); + } } 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..511aa1db 100644 --- a/src/Grpc/orchestrator_service.proto +++ b/src/Grpc/orchestrator_service.proto @@ -823,6 +823,31 @@ 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); + + // Drains blob-externalized payload tombstones from the backend so the worker can delete the + // corresponding blobs from customer storage (the backend has no storage credentials of its own). + // This is a push model: the backend streams TombstonedPayload messages; the worker deletes each + // blob and acknowledges with a PayloadPurgeAck message on the same stream, after which the backend + // hard-deletes the soft-deleted row. The stream stays open for the worker's lifetime. + rpc PurgeExternalPayloads(stream PayloadPurgeAck) returns (stream TombstonedPayload); +} + +// 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; } message GetWorkItemsRequest { diff --git a/test/AzureBlobPayloads.Tests/AutoPurge/BlobPayloadAutoPurgeServiceTests.cs b/test/AzureBlobPayloads.Tests/AutoPurge/BlobPayloadAutoPurgeServiceTests.cs new file mode 100644 index 00000000..834164ad --- /dev/null +++ b/test/AzureBlobPayloads.Tests/AutoPurge/BlobPayloadAutoPurgeServiceTests.cs @@ -0,0 +1,78 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Xunit; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests; + +public class BlobPayloadAutoPurgeServiceTests +{ + [Fact] + public void Constructor_FromCallInvokerAndStore_ProducesHostedServiceWithoutDependencyInjection() + { + // Arrange (hosts that don't use UseExternalizedPayloads reuse the service through this constructor) + CallInvoker callInvoker = Mock.Of(); + PayloadStore store = Mock.Of(); + + // Act + BlobPayloadAutoPurgeService service = new(callInvoker, store); + + // Assert + Assert.IsAssignableFrom(service); + } + + [Fact] + public void Constructor_FromChannelAndStore_ProducesHostedService() + { + // Arrange + using GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:9999"); + PayloadStore store = Mock.Of(); + + // Act + BlobPayloadAutoPurgeService service = new(channel, store); + + // Assert + Assert.IsAssignableFrom(service); + } + + [Fact] + public void Constructor_AcceptsPlainILogger() + { + // Arrange (a reusing host can pass any ILogger, not only ILogger) + CallInvoker callInvoker = Mock.Of(); + PayloadStore store = Mock.Of(); + ILogger logger = NullLogger.Instance; + + // Act + BlobPayloadAutoPurgeService service = new(callInvoker, store, logger); + + // Assert + Assert.IsAssignableFrom(service); + } + + [Fact] + public void Constructor_NullCallInvoker_Throws() + { + // Arrange + PayloadStore store = Mock.Of(); + + // Act & Assert + Assert.Throws(() => new BlobPayloadAutoPurgeService((CallInvoker)null!, store)); + } + + [Fact] + public void Constructor_NullStore_Throws() + { + // Arrange + CallInvoker callInvoker = Mock.Of(); + + // Act & Assert + Assert.Throws(() => new BlobPayloadAutoPurgeService(callInvoker, null!)); + } +} diff --git a/test/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj b/test/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj new file mode 100644 index 00000000..5fc664cf --- /dev/null +++ b/test/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj @@ -0,0 +1,23 @@ + + + + net10.0 + enable + enable + false + true + + Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests + Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests + + + + + + + + + + + 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)); + } +}