diff --git a/Microsoft.DurableTask.sln b/Microsoft.DurableTask.sln index 3c6d4539..380ee0a0 100644 --- a/Microsoft.DurableTask.sln +++ b/Microsoft.DurableTask.sln @@ -145,6 +145,10 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "AzureManaged", "AzureManage EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Grpc", "Grpc", "{3B8F957E-7773-4C0C-ACD7-91A1591D9312}" EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Extensions", "Extensions", "{00205C88-F000-28F2-A910-C6FA00E065EE}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "AzureBlobPayloads.Tests", "test\Extensions\AzureBlobPayloads.Tests\AzureBlobPayloads.Tests.csproj", "{3E509481-3CCC-4006-BCB2-9E8FA7C275F1}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -839,6 +843,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 + {3E509481-3CCC-4006-BCB2-9E8FA7C275F1}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {3E509481-3CCC-4006-BCB2-9E8FA7C275F1}.Debug|Any CPU.Build.0 = Debug|Any CPU + {3E509481-3CCC-4006-BCB2-9E8FA7C275F1}.Debug|x64.ActiveCfg = Debug|Any CPU + {3E509481-3CCC-4006-BCB2-9E8FA7C275F1}.Debug|x64.Build.0 = Debug|Any CPU + {3E509481-3CCC-4006-BCB2-9E8FA7C275F1}.Debug|x86.ActiveCfg = Debug|Any CPU + {3E509481-3CCC-4006-BCB2-9E8FA7C275F1}.Debug|x86.Build.0 = Debug|Any CPU + {3E509481-3CCC-4006-BCB2-9E8FA7C275F1}.Release|Any CPU.ActiveCfg = Release|Any CPU + {3E509481-3CCC-4006-BCB2-9E8FA7C275F1}.Release|Any CPU.Build.0 = Release|Any CPU + {3E509481-3CCC-4006-BCB2-9E8FA7C275F1}.Release|x64.ActiveCfg = Release|Any CPU + {3E509481-3CCC-4006-BCB2-9E8FA7C275F1}.Release|x64.Build.0 = Release|Any CPU + {3E509481-3CCC-4006-BCB2-9E8FA7C275F1}.Release|x86.ActiveCfg = Release|Any CPU + {3E509481-3CCC-4006-BCB2-9E8FA7C275F1}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -911,6 +927,8 @@ 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} + {00205C88-F000-28F2-A910-C6FA00E065EE} = {E5637F81-2FB9-4CD7-900D-455363B142A7} + {3E509481-3CCC-4006-BCB2-9E8FA7C275F1} = {00205C88-F000-28F2-A910-C6FA00E065EE} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {AB41CB55-35EA-4986-A522-387AB3402E71} diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs index c8d59eec..f519b196 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -17,6 +17,10 @@ namespace Microsoft.DurableTask; /// blob:v2:{fullBlobUrl}, where the URL is the blob's absolute URI including the storage account. /// Legacy blob:v1:{container}:{blobName} tokens are still recognized for read back-compatibility. /// +[System.Diagnostics.CodeAnalysis.SuppressMessage( + "Design", + "CA1001:Types that own disposable fields should be disposable", + Justification = "SemaphoreSlim does not allocate a disposable resource unless AvailableWaitHandle is accessed.")] public sealed class BlobPayloadStore : PayloadStore { const string TokenPrefixV1 = "blob:v1:"; @@ -26,9 +30,15 @@ public sealed class BlobPayloadStore : PayloadStore const int BaseDelayMs = 250; const int MaxDelayMs = 10_000; const int NetworkTimeoutMinutes = 2; + readonly BlobContainerClient containerClient; readonly LargePayloadStorageOptions options; readonly BlobClientOptions clientOptions; + readonly SemaphoreSlim containerInitializationLock = new(initialCount: 1, maxCount: 1); + + // Each successful initialization publishes a unique token. The token lets a stale + // ContainerNotFound failure invalidate only the initialization generation it used. + object? containerGeneration; /// /// Initializes a new instance of the class. @@ -71,6 +81,19 @@ public BlobPayloadStore(LargePayloadStorageOptions options) this.containerClient = serviceClient.GetBlobContainerClient(options.ContainerName); } + /// + /// Initializes a new instance of the class using an existing + /// container client. Intended for unit testing only. + /// + /// The options for the blob payload store. + /// The blob container client to use. + internal BlobPayloadStore(LargePayloadStorageOptions options, BlobContainerClient containerClient) + { + this.options = options ?? throw new ArgumentNullException(nameof(options)); + this.containerClient = containerClient ?? throw new ArgumentNullException(nameof(containerClient)); + this.clientOptions = new BlobClientOptions(); + } + /// public override async Task UploadAsync(string payLoad, CancellationToken cancellationToken) { @@ -80,36 +103,58 @@ public override async Task UploadAsync(string payLoad, CancellationToken byte[] payloadBuffer = Encoding.UTF8.GetBytes(payLoad); - // Ensure container exists (idempotent) - await this.containerClient.CreateIfNotExistsAsync(PublicAccessType.None, default, default, cancellationToken); - - if (this.options.CompressionEnabled) + // Ensure container exists. Cached/single-flight after the first successful call so we + // don't pay for an extra CreateIfNotExistsAsync request/transaction on every upload. + // Retry one write after an out-of-band container deletion so the cached path preserves + // the recovery behavior that an unconditional CreateIfNotExistsAsync provided before + // initialization was cached. + bool retryAfterContainerNotFound = true; + while (true) { - BlobOpenWriteOptions writeOptions = new() - { - HttpHeaders = new BlobHttpHeaders { ContentEncoding = ContentEncodingGzip }, - }; - using Stream blobStream = await blob.OpenWriteAsync(true, writeOptions, cancellationToken); - using GZipStream compressedBlobStream = new(blobStream, System.IO.Compression.CompressionLevel.Optimal, leaveOpen: true); - - // using MemoryStream payloadStream = new(payloadBuffer, writable: false); + object generation = await this.EnsureContainerExistsAsync(cancellationToken).ConfigureAwait(false); - // await payloadStream.CopyToAsync(compressedBlobStream, bufferSize: DefaultCopyBufferSize, cancellationToken); - await WritePayloadAsync(payloadBuffer, compressedBlobStream, cancellationToken); - await compressedBlobStream.FlushAsync(cancellationToken); - await blobStream.FlushAsync(cancellationToken); - } - else - { - using Stream blobStream = await blob.OpenWriteAsync(true, default, cancellationToken); + try + { + if (this.options.CompressionEnabled) + { + BlobOpenWriteOptions writeOptions = new() + { + HttpHeaders = new BlobHttpHeaders { ContentEncoding = ContentEncodingGzip }, + }; + using Stream blobStream = await blob.OpenWriteAsync(true, writeOptions, cancellationToken); + using GZipStream compressedBlobStream = new(blobStream, System.IO.Compression.CompressionLevel.Optimal, leaveOpen: true); + + // using MemoryStream payloadStream = new(payloadBuffer, writable: false); + + // await payloadStream.CopyToAsync(compressedBlobStream, bufferSize: DefaultCopyBufferSize, cancellationToken); + await WritePayloadAsync(payloadBuffer, compressedBlobStream, cancellationToken); + await compressedBlobStream.FlushAsync(cancellationToken); + await blobStream.FlushAsync(cancellationToken); + } + else + { + using Stream blobStream = await blob.OpenWriteAsync(true, default, cancellationToken); + + // using MemoryStream payloadStream = new(payloadBuffer, writable: false); + // await payloadStream.CopyToAsync(blobStream, bufferSize: DefaultCopyBufferSize, cancellationToken); + await WritePayloadAsync(payloadBuffer, blobStream, cancellationToken); + await blobStream.FlushAsync(cancellationToken); + } + } + catch (RequestFailedException ex) when ( + retryAfterContainerNotFound && + ex.ErrorCode == BlobErrorCode.ContainerNotFound) + { + // Clear only the generation used by this upload, then retry once. If Azure is still + // deleting the container, CreateIfNotExistsAsync can surface ContainerBeingDeleted, + // matching the behavior before initialization was cached. + _ = Interlocked.CompareExchange(ref this.containerGeneration, null, generation); + retryAfterContainerNotFound = false; + continue; + } - // using MemoryStream payloadStream = new(payloadBuffer, writable: false); - // await payloadStream.CopyToAsync(blobStream, bufferSize: DefaultCopyBufferSize, cancellationToken); - await WritePayloadAsync(payloadBuffer, blobStream, cancellationToken); - await blobStream.FlushAsync(cancellationToken); + return EncodeToken(blob.Uri); } - - return EncodeToken(blob.Uri); } /// @@ -277,6 +322,39 @@ bool IsConfiguredContainer(Uri tokenContainerUri) StringComparison.Ordinal); } + /// + /// Ensures the container exists and returns the initialization generation used by the caller. + /// + async Task EnsureContainerExistsAsync(CancellationToken cancellationToken) + { + object? generation = Volatile.Read(ref this.containerGeneration); + if (generation is not null) + { + return generation; + } + + await this.containerInitializationLock.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + generation = Volatile.Read(ref this.containerGeneration); + if (generation is not null) + { + return generation; + } + + await this.containerClient.CreateIfNotExistsAsync(PublicAccessType.None, cancellationToken: cancellationToken) + .ConfigureAwait(false); + + generation = new object(); + Volatile.Write(ref this.containerGeneration, generation); + return generation; + } + finally + { + this.containerInitializationLock.Release(); + } + } + /// /// The result of decoding an externalized payload token. /// diff --git a/test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj b/test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj new file mode 100644 index 00000000..39298f69 --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj @@ -0,0 +1,14 @@ + + + + net10.0 + + Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests + $(AssemblyName) + + + + + + + diff --git a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs new file mode 100644 index 00000000..c03ee76b --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs @@ -0,0 +1,393 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Azure; +using Azure.Storage.Blobs; +using Azure.Storage.Blobs.Models; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests; + +/// +/// Unit tests for 's container-initialization caching behavior +/// (see https://github.com/microsoft/durabletask-dotnet/issues/771). +/// +public class BlobPayloadStoreTests +{ + [Fact] + public async Task UploadAsync_ConcurrentCallers_OnlyCreatesContainerOnce() + { + // Arrange + const int WorkerCount = 16; + int createCalls = 0; + Mock containerClientMock = CreateContainerClientMock(); + containerClientMock + .Setup(c => c.CreateIfNotExistsAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .Returns(async () => + { + Interlocked.Increment(ref createCalls); + await Task.Delay(50); + return (Response)null!; + }); + + BlobPayloadStore store = new(new LargePayloadStorageOptions(), containerClientMock.Object); + + // Act: release all workers from an async gate after each has reached it. The delayed + // create keeps initialization in flight while the workers contend for the gate. + TaskCompletionSource startGate = new(TaskCreationOptions.RunContinuationsAsynchronously); + int readyWorkers = 0; + Task[] uploads = Enumerable.Range(0, WorkerCount) + .Select(_ => Task.Run(async () => + { + if (Interlocked.Increment(ref readyWorkers) == WorkerCount) + { + startGate.SetResult(true); + } + + await startGate.Task; + return await store.UploadAsync("payload", CancellationToken.None); + })) + .ToArray(); + await Task.WhenAll(uploads); + + // Assert + createCalls.Should().Be(1); + containerClientMock.Verify( + c => c.CreateIfNotExistsAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Once); + } + + [Fact] + public async Task UploadAsync_FailedInitialization_IsRetriedOnNextCall() + { + // Arrange: the first CreateIfNotExistsAsync attempt fails (e.g. transient storage error), + // the second succeeds. + Mock containerClientMock = CreateContainerClientMock(); + containerClientMock + .SetupSequence(c => c.CreateIfNotExistsAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .ThrowsAsync(new RequestFailedException(503, "Service unavailable")) + .ReturnsAsync((Response)null!); + + BlobPayloadStore store = new(new LargePayloadStorageOptions(), containerClientMock.Object); + + // Act + Func firstUpload = () => store.UploadAsync("payload", CancellationToken.None); + string secondToken; + await firstUpload.Should().ThrowAsync(); + secondToken = await store.UploadAsync("payload", CancellationToken.None); + + // Assert: the failed attempt was not cached, so the second call retried initialization. + secondToken.Should().NotBeNullOrEmpty(); + containerClientMock.Verify( + c => c.CreateIfNotExistsAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Exactly(2)); + } + + [Fact] + public async Task UploadAsync_CancelledInitializer_AllowsWaitingCallerToRetry() + { + // Arrange: the first caller owns initialization until its token is cancelled. A waiting + // caller must then acquire the gate and retry with its own uncancelled token. + TaskCompletionSource firstCreateStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); + int createCalls = 0; + Mock containerClientMock = CreateContainerClientMock(); + containerClientMock + .Setup(c => c.CreateIfNotExistsAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .Returns(( + PublicAccessType _, + IDictionary _, + BlobContainerEncryptionScopeOptions _, + CancellationToken cancellationToken) => CreateContainerAsync(cancellationToken)); + + BlobPayloadStore store = new(new LargePayloadStorageOptions(), containerClientMock.Object); + + using CancellationTokenSource cts = new(); + Task cancelledUpload = store.UploadAsync("payload", cts.Token); + await firstCreateStarted.Task; + Task otherUpload = store.UploadAsync("payload", CancellationToken.None); + + // Act + cts.Cancel(); + + // Assert + Task completedTask = await Task.WhenAny(cancelledUpload, Task.Delay(TimeSpan.FromSeconds(5))); + completedTask.Should().BeSameAs(cancelledUpload, "the caller token should cancel the storage initialization"); + Func awaitCancelled = () => cancelledUpload; + await awaitCancelled.Should().ThrowAsync(); + + string token = await otherUpload; + token.Should().NotBeNullOrEmpty(); + containerClientMock.Verify( + c => c.CreateIfNotExistsAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Exactly(2)); + + async Task> CreateContainerAsync(CancellationToken cancellationToken) + { + if (Interlocked.Increment(ref createCalls) == 1) + { + firstCreateStarted.SetResult(true); + await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); + } + + return null!; + } + } + + [Fact] + public async Task UploadAsync_CancelledWaiter_DoesNotCancelInitializer() + { + // Arrange: hold the first caller in initialization while a second caller waits for the + // gate with an independently cancellable token. + TaskCompletionSource> initTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource createStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); + Mock containerClientMock = CreateContainerClientMock(); + containerClientMock + .Setup(c => c.CreateIfNotExistsAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .Returns(() => + { + createStarted.SetResult(true); + return initTcs.Task; + }); + + BlobPayloadStore store = new(new LargePayloadStorageOptions(), containerClientMock.Object); + Task initializingUpload = store.UploadAsync("payload", CancellationToken.None); + await createStarted.Task; + + using CancellationTokenSource cts = new(); + Task waitingUpload = store.UploadAsync("payload", cts.Token); + + // Act + cts.Cancel(); + + // Assert + Task completedTask = await Task.WhenAny(waitingUpload, Task.Delay(TimeSpan.FromSeconds(5))); + completedTask.Should().BeSameAs(waitingUpload, "the caller token should cancel the wait for initialization"); + Func awaitWaitingUpload = () => waitingUpload; + await awaitWaitingUpload.Should().ThrowAsync(); + + initTcs.SetResult(null!); + string token = await initializingUpload; + token.Should().NotBeNullOrEmpty(); + containerClientMock.Verify( + c => c.CreateIfNotExistsAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Once); + } + + [Fact] + public async Task UploadAsync_ContainerDeletedAfterInitializationAndDeletionSettles_RecreatesContainer() + { + // Arrange: the first write observes an out-of-band deletion, and the mocked re-create + // succeeds as it would after Azure has finished deleting the container. While deletion is + // still in progress, Azure can return ContainerBeingDeleted and that error propagates. + Mock containerClientMock = CreateContainerClientMock(); + containerClientMock + .Setup(c => c.CreateIfNotExistsAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .ReturnsAsync((Response)null!); + + Mock blobClientMock = new(); + blobClientMock.SetupGet(b => b.Uri).Returns(new Uri("https://testaccount.blob.core.windows.net/test-container/payload")); + using MemoryStream successfulWriteStream = new(); + blobClientMock + .SetupSequence(b => b.OpenWriteAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new RequestFailedException(404, "The specified container does not exist.", BlobErrorCode.ContainerNotFound.ToString(), null)) + .ReturnsAsync(successfulWriteStream); + containerClientMock.Setup(c => c.GetBlobClient(It.IsAny())).Returns(blobClientMock.Object); + + BlobPayloadStore store = new(new LargePayloadStorageOptions(), containerClientMock.Object); + + // Act + string token = await store.UploadAsync("payload", CancellationToken.None); + + // Assert: the missing-container failure reset the cached generation and this same upload + // re-created the container after deletion had settled. + token.Should().NotBeNullOrEmpty(); + containerClientMock.Verify( + c => c.CreateIfNotExistsAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Exactly(2)); + } + + [Fact] + public async Task UploadAsync_UnrelatedStorageError_PropagatesWithoutInvalidatingCache() + { + // Arrange: initialization succeeds; the first blob write fails with an error that has + // nothing to do with the container being missing (should not trigger re-initialization). + Mock containerClientMock = CreateContainerClientMock(); + containerClientMock + .Setup(c => c.CreateIfNotExistsAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .ReturnsAsync((Response)null!); + + Mock blobClientMock = new(); + blobClientMock.SetupGet(b => b.Uri).Returns(new Uri("https://testaccount.blob.core.windows.net/test-container/payload")); + using MemoryStream retryWriteStream = new(); + blobClientMock + .SetupSequence(b => b.OpenWriteAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new RequestFailedException(500, "Internal server error")) + .ReturnsAsync(retryWriteStream); + containerClientMock.Setup(c => c.GetBlobClient(It.IsAny())).Returns(blobClientMock.Object); + + BlobPayloadStore store = new(new LargePayloadStorageOptions(), containerClientMock.Object); + + // Act + Func firstUpload = () => store.UploadAsync("payload", CancellationToken.None); + await firstUpload.Should().ThrowAsync(); + + string secondToken = await store.UploadAsync("payload", CancellationToken.None); + + // Assert: the unrelated error propagated to the caller, and initialization was not + // repeated since the cached container state was still considered valid. + secondToken.Should().NotBeNullOrEmpty(); + containerClientMock.Verify( + c => c.CreateIfNotExistsAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Once); + } + + [Fact] + public async Task UploadAsync_StaleContainerNotFoundFailure_DoesNotOverwriteNewerInitializer() + { + // Arrange: container creation always succeeds when invoked. + int createCalls = 0; + Mock containerClientMock = new(); + containerClientMock.Setup(c => c.Name).Returns("test-container"); + containerClientMock + .Setup(c => c.CreateIfNotExistsAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .Returns(() => + { + Interlocked.Increment(ref createCalls); + return Task.FromResult((Response)null!); + }); + + // The first upload's blob write is held open (via this TCS) until the test explicitly + // releases it, simulating a slow write that only discovers the container is gone after + // other, faster uploads have already observed the deletion and re-initialized. + TaskCompletionSource firstUploadWriteTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); + + int getBlobClientCalls = 0; + containerClientMock + .Setup(c => c.GetBlobClient(It.IsAny())) + .Returns(() => + { + int callIndex = Interlocked.Increment(ref getBlobClientCalls); + Mock blobClientMock = new(); + blobClientMock.SetupGet(b => b.Uri).Returns(new Uri($"https://testaccount.blob.core.windows.net/test-container/payload-{callIndex}")); + switch (callIndex) + { + case 1: + // First upload: its write stays pending until the test releases it. + blobClientMock + .Setup(b => b.OpenWriteAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .Returns(firstUploadWriteTcs.Task); + break; + case 2: + // Second upload: discovers the container is gone immediately. + blobClientMock + .Setup(b => b.OpenWriteAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new RequestFailedException(404, "The specified container does not exist.", BlobErrorCode.ContainerNotFound.ToString(), null)); + break; + default: + blobClientMock + .Setup(b => b.OpenWriteAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(() => new MemoryStream()); + break; + } + + return blobClientMock.Object; + }); + + BlobPayloadStore store = new(new LargePayloadStorageOptions(), containerClientMock.Object); + + // Act + // Upload 1 initializes the container (create call #1) and then blocks mid-write. + Task upload1 = store.UploadAsync("payload", CancellationToken.None); + + // Upload 2 reuses the already-cached initializer (no new create call), fails on write + // with ContainerNotFound, and invalidates that initializer. + Func upload2 = () => store.UploadAsync("payload", CancellationToken.None); + await upload2.Should().ThrowAsync(); + + // Upload 3 observes the invalidated cache and re-initializes the container (create call + // #2), publishing a newer initializer. + string upload3Token = await store.UploadAsync("payload", CancellationToken.None); + upload3Token.Should().NotBeNullOrEmpty(); + + // Now let upload 1's stale write finally fail with the same ContainerNotFound error. Its + // cache invalidation must not clobber the newer initializer upload 3 already published. + firstUploadWriteTcs.SetException(new RequestFailedException(404, "The specified container does not exist.", BlobErrorCode.ContainerNotFound.ToString(), null)); + Func awaitUpload1 = () => upload1; + await awaitUpload1.Should().ThrowAsync(); + + // Assert: a further upload reuses upload 3's initializer instead of triggering a third, + // unnecessary re-initialization caused by upload 1's stale failure. + string finalToken = await store.UploadAsync("payload", CancellationToken.None); + finalToken.Should().NotBeNullOrEmpty(); + createCalls.Should().Be(2); + } + + static Mock CreateContainerClientMock() + { + Mock containerClientMock = new(); + containerClientMock.Setup(c => c.Name).Returns("test-container"); + containerClientMock + .Setup(c => c.GetBlobClient(It.IsAny())) + .Returns(() => + { + Mock blobClientMock = new(); + blobClientMock.SetupGet(b => b.Uri).Returns(new Uri("https://testaccount.blob.core.windows.net/test-container/payload")); + blobClientMock + .Setup(b => b.OpenWriteAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(() => new MemoryStream()); + return blobClientMock.Object; + }); + return containerClientMock; + } +} diff --git a/test/Extensions/AzureBlobPayloads.Tests/Usings.cs b/test/Extensions/AzureBlobPayloads.Tests/Usings.cs new file mode 100644 index 00000000..39fe94e9 --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/Usings.cs @@ -0,0 +1,6 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +global using FluentAssertions; +global using Moq; +global using Xunit;