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 e01d2e74..76a7a7c7 100644
--- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs
+++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs
@@ -23,9 +23,33 @@ public sealed class BlobPayloadStore : PayloadStore
const int BaseDelayMs = 250;
const int MaxDelayMs = 10_000;
const int NetworkTimeoutMinutes = 2;
+
+ ///
+ /// A no-op fault-observer used as the default value of ,
+ /// so that field is never and
+ /// can invoke it unconditionally - see the remarks on
+ /// for why this matters.
+ ///
+ static readonly Action NoOpFaultObserver = static _ => { };
+
readonly BlobContainerClient containerClient;
readonly LargePayloadStorageOptions options;
+ // Caches the single in-flight (or completed) container-initialization gate so that
+ // concurrent/subsequent uploads don't each issue their own CreateIfNotExistsAsync request.
+ // Null means "not yet attempted" or "needs to be retried". The Lazy wrapper makes
+ // initialization truly single-flight: publishing the gate (a cheap CompareExchange) always
+ // happens before any real work starts, and Lazy guarantees the factory (which starts the
+ // real CreateIfNotExistsAsync call) runs exactly once even if multiple callers race to
+ // access Value concurrently. See PublishNewInitializer and CreateContainerIfNotExistsAsync.
+ Lazy? containerInitializer;
+
+ ///
+ /// Backing field for . Never ;
+ /// defaults to, and is reset back to, .
+ ///
+ Action onInitializationFaultObserved = NoOpFaultObserver;
+
///
/// Initializes a new instance of the class.
///
@@ -67,6 +91,55 @@ 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));
+ }
+
+ ///
+ /// Gets or sets a test-only hook, invoked exactly once per initialization attempt that
+ /// ultimately faults, immediately after the fault-observing continuation attached in
+ /// reads the shared initializer's
+ /// (see ). This
+ /// lets unit tests deterministically verify the continuation actually ran and observed the
+ /// fault, instead of relying on plus
+ /// forced garbage collection - which is sensitive to GC/finalization timing, debugger
+ /// attachment, and JIT optimizations, and so cannot reliably distinguish "the fix ran" from
+ /// "the CLR just hasn't collected the task yet".
+ ///
+ /// This property is never : it defaults to a no-op delegate, and
+ /// setting it to resets it back to that no-op rather than making the
+ /// backing field nullable. This is deliberate, not just a convenience - it lets
+ /// invoke this hook directly, with no
+ /// null-conditional operator, which in turn means the same statement runs whether a test has
+ /// overridden this hook or not. That eliminates - by construction, not merely by test
+ /// coverage - the historical bug where the continuation used
+ /// this.OnInitializationFaultObserved?.Invoke(t.Exception!): because ?.Invoke(...)
+ /// short-circuits and skips evaluating its argument when the target is ,
+ /// that pattern silently never read in production (where the hook
+ /// was by default), even though every test that exercised it happened to
+ /// set a non-null hook first and so could never observe the regression. With this hook
+ /// guaranteed non-null, that class of bug cannot recur regardless of how the invocation is
+ /// written at the call site, and a test that overrides this hook is exercising the exact same
+ /// code as the unmodified production default - just substituting a different delegate
+ /// instance for .
+ ///
+ /// Each test uses its own instance, so no reset between tests
+ /// is needed. Callers of this hook must not assume any particular thread.
+ ///
+ internal Action OnInitializationFaultObserved
+ {
+ get => this.onInitializationFaultObserved;
+ set => this.onInitializationFaultObserved = value ?? NoOpFaultObserver;
+ }
+
///
public override async Task UploadAsync(string payLoad, CancellationToken cancellationToken)
{
@@ -76,36 +149,63 @@ 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()
+ // Keep the specific initializer instance this upload used so ContainerNotFound
+ // recovery below can invalidate it precisely (see the catch block).
+ Lazy containerInitializer = await this.EnsureContainerExistsAsync(cancellationToken).ConfigureAwait(false);
+
+ try
{
- 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);
+ 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);
+ // 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);
+ // 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);
- }
+ // 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)
+ {
+ // The container existed when we last verified/created it but has since been deleted
+ // (e.g. by an operator). Clear the cached initializer so this same upload can recreate
+ // the container and retry once. CompareExchange against the specific initializer this
+ // attempt used ensures a stale failure can never clobber a newer initializer already
+ // published by another, faster-recovering concurrent upload that detected the same
+ // deletion.
+ _ = Interlocked.CompareExchange(ref this.containerInitializer, null, containerInitializer);
+ retryAfterContainerNotFound = false;
+ continue;
+ }
- return EncodeToken(this.containerClient.Name, blobName);
+ return EncodeToken(this.containerClient.Name, blobName);
+ }
}
///
@@ -156,6 +256,35 @@ public override bool IsKnownPayloadToken(string value)
return value.StartsWith(TokenPrefix, StringComparison.Ordinal);
}
+ ///
+ /// Awaits a shared initialization task while still honoring the caller's own cancellation
+ /// token, without cancelling the shared task itself.
+ ///
+ static async Task WaitForInitializationAsync(Task initializationTask, CancellationToken cancellationToken)
+ {
+#if NETSTANDARD2_0
+ if (!cancellationToken.CanBeCanceled || initializationTask.IsCompleted)
+ {
+ await initializationTask.ConfigureAwait(false);
+ return;
+ }
+
+ TaskCompletionSource cancellationTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ using (cancellationToken.Register(static state => ((TaskCompletionSource)state!).TrySetResult(true), cancellationTcs))
+ {
+ Task completed = await Task.WhenAny(initializationTask, cancellationTcs.Task).ConfigureAwait(false);
+ if (completed == cancellationTcs.Task)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+ }
+ }
+
+ await initializationTask.ConfigureAwait(false);
+#else
+ await initializationTask.WaitAsync(cancellationToken).ConfigureAwait(false);
+#endif
+ }
+
static async Task WritePayloadAsync(byte[] payloadBuffer, Stream target, CancellationToken cancellationToken)
{
#if NETSTANDARD2_0
@@ -195,4 +324,102 @@ static async Task ReadToEndAsync(StreamReader reader, CancellationToken
return (rest.Substring(0, sep), rest.Substring(sep + 1));
}
+
+ ///
+ /// Ensures the container exists, issuing at most one CreateIfNotExistsAsync request
+ /// across all concurrent/subsequent callers, and returns the specific initializer gate that
+ /// was used. The result is cached for the lifetime of this instance once it completes
+ /// successfully; a failed attempt self-heals (see )
+ /// so the next caller retries instead of observing a stale failure forever.
+ ///
+ async Task> EnsureContainerExistsAsync(CancellationToken cancellationToken)
+ {
+ Lazy initializer = Volatile.Read(ref this.containerInitializer)
+ ?? this.PublishNewInitializer();
+
+ await WaitForInitializationAsync(initializer.Value, cancellationToken).ConfigureAwait(false);
+ return initializer;
+ }
+
+ ///
+ /// Atomically publishes a not-yet-started initialization gate, in a true single-flight
+ /// manner: publishing the gate (a cheap compare-exchange)
+ /// always completes before any real work starts, so racing first-time callers can never
+ /// each independently trigger their own CreateIfNotExistsAsync request. Only the
+ /// single winning gate is stored, and (with
+ /// ) guarantees its factory -
+ /// which starts the real container-creation call - is invoked exactly once even when
+ /// multiple callers race to access concurrently. The winning
+ /// publisher also attaches a single fault-observer (see
+ /// ) to the shared task, once, up front -
+ /// rather than leaving each individual caller responsible for observing failures on its own
+ /// cancellation path - so a shared initialization failure is always observed regardless of
+ /// how many (if any) callers are still waiting on it when it faults, on every target
+ /// framework this library supports.
+ ///
+ Lazy PublishNewInitializer()
+ {
+ Lazy? initializer = null;
+ initializer = new Lazy(
+ () => this.CreateContainerIfNotExistsAsync(initializer!),
+ LazyThreadSafetyMode.ExecutionAndPublication);
+
+ Lazy published = Interlocked.CompareExchange(ref this.containerInitializer, initializer, null) ?? initializer;
+ if (ReferenceEquals(published, initializer))
+ {
+ this.ObserveFaultWithoutAwaiting(published.Value);
+ }
+
+ return published;
+ }
+
+ ///
+ /// Attaches a fire-and-forget continuation that reads if
+ /// ultimately faults, marking that fault as "observed" without
+ /// awaiting or blocking on the task. Used so a shared task's eventual failure is always
+ /// observed even if every caller that was waiting on it stops doing so (e.g. because each
+ /// caller's own fired first) - otherwise the runtime would
+ /// report it via once the task is
+ /// garbage-collected. Also invokes (a no-op in
+ /// production) so tests can deterministically confirm this continuation ran - it is invoked
+ /// directly, with no null-conditional operator, because it is guaranteed non-null (see its
+ /// remarks); this is what makes reading here unconditional in
+ /// every configuration, not merely in the current source form of this method.
+ ///
+ void ObserveFaultWithoutAwaiting(Task task)
+ {
+ _ = task.ContinueWith(
+ t => this.OnInitializationFaultObserved(t.Exception!),
+ CancellationToken.None,
+ TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
+ TaskScheduler.Default);
+ }
+
+ ///
+ /// Creates the container if it doesn't already exist. This task may be shared by many
+ /// concurrent callers, each with its own independently-cancellable ;
+ /// it intentionally does not use any single caller's token (see
+ /// , which lets individual
+ /// callers stop waiting without cancelling the shared operation for everyone else).
+ /// If the underlying call fails, this method self-heals the cache by clearing it as part of
+ /// its own completion - independent of whether any particular caller is still around to
+ /// observe the failure - so the next upload gets a fresh initialization attempt instead of a
+ /// stale error. The CompareExchange only clears the cache if it still points at this same
+ /// gate (), so this can never erase a newer initializer already
+ /// published by a racing caller (e.g. one that recovered from a concurrently-deleted
+ /// container) after this attempt started.
+ ///
+ async Task CreateContainerIfNotExistsAsync(Lazy self)
+ {
+ try
+ {
+ await this.containerClient.CreateIfNotExistsAsync(PublicAccessType.None, cancellationToken: CancellationToken.None)
+ .ConfigureAwait(false);
+ }
+ catch
+ {
+ _ = Interlocked.CompareExchange(ref this.containerInitializer, null, self);
+ throw;
+ }
+ }
}
diff --git a/test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj b/test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj
new file mode 100644
index 00000000..ece939f8
--- /dev/null
+++ b/test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj
@@ -0,0 +1,16 @@
+
+
+
+ net10.0
+ Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests
+
+
+
+
+
+
+
+
+
+
+
diff --git a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs
new file mode 100644
index 00000000..5c8576c2
--- /dev/null
+++ b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs
@@ -0,0 +1,510 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using Azure;
+using Azure.Storage.Blobs;
+using Azure.Storage.Blobs.Models;
+using Microsoft.DurableTask;
+
+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: force genuinely concurrent entry into the single-flight logic. Each worker runs
+ // on its own thread-pool thread (via Task.Run) and blocks at a Barrier until every
+ // worker has arrived, so they all call UploadAsync at (as close to) the same instant as
+ // possible. A simple sequential LINQ + Task.WhenAll loop would not exercise this: the
+ // synchronous prefix of each call (up to the first real await) would run one after
+ // another on the calling thread, letting the first caller publish the cached
+ // initializer before any other caller's code even starts, which trivially "passes" even
+ // a buggy, non-single-flight implementation.
+ using Barrier barrier = new(WorkerCount);
+ Task[] uploads = Enumerable.Range(0, WorkerCount)
+ .Select(_ => Task.Run(async () =>
+ {
+ barrier.SignalAndWait();
+ 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_CancelledCaller_DoesNotFaultSharedInitializationForOtherCallers()
+ {
+ // Arrange: control exactly when the shared initialization completes.
+ TaskCompletionSource> initTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ Mock containerClientMock = CreateContainerClientMock();
+ containerClientMock
+ .Setup(c => c.CreateIfNotExistsAsync(
+ It.IsAny(),
+ It.IsAny>(),
+ It.IsAny(),
+ It.IsAny()))
+ .Returns(initTcs.Task);
+
+ BlobPayloadStore store = new(new LargePayloadStorageOptions(), containerClientMock.Object);
+
+ using CancellationTokenSource cts = new();
+ Task cancelledUpload = store.UploadAsync("payload", cts.Token);
+ Task otherUpload = store.UploadAsync("payload", CancellationToken.None);
+
+ // Act: cancel the first caller while initialization is still in flight, then let
+ // initialization complete successfully for everyone else.
+ cts.Cancel();
+ await Task.Delay(50);
+ initTcs.SetResult(null!);
+
+ // Assert
+ 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.Once);
+ }
+
+ [Fact]
+ public async Task UploadAsync_ContainerDeletedAfterInitialization_ResetsCacheAndRecreatesContainer()
+ {
+ // Arrange: initialization always succeeds (from the SDK's point of view), but the first
+ // blob write fails because the container was deleted out-of-band after initialization.
+ Mock containerClientMock = CreateContainerClientMock();
+ containerClientMock
+ .Setup(c => c.CreateIfNotExistsAsync(
+ It.IsAny(),
+ It.IsAny>(),
+ It.IsAny(),
+ It.IsAny()))
+ .ReturnsAsync((Response)null!);
+
+ Mock blobClientMock = new();
+ 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 container-not-found failure reset the cache and this same upload recreated
+ // the container instead of exposing a transient recovery detail to the caller.
+ 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();
+ 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_AllWaitersCancelBeforeInitializationFails_NextUploadRetriesWithFreshAttempt()
+ {
+ // Arrange: control exactly when the shared initialization completes/fails. The first
+ // invocation returns this controlled (eventually-faulted) task; any subsequent
+ // invocation - i.e. the fresh retry we're testing for - succeeds immediately.
+ TaskCompletionSource> initTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ Mock containerClientMock = CreateContainerClientMock();
+ containerClientMock
+ .SetupSequence(c => c.CreateIfNotExistsAsync(
+ It.IsAny(),
+ It.IsAny>(),
+ It.IsAny(),
+ It.IsAny()))
+ .Returns(initTcs.Task)
+ .ReturnsAsync((Response)null!);
+
+ BlobPayloadStore store = new(new LargePayloadStorageOptions(), containerClientMock.Object);
+
+ // CreateContainerIfNotExistsAsync clears the cache (Interlocked.CompareExchange) in its
+ // catch block strictly before re-throwing, and OnInitializationFaultObserved is invoked
+ // from a continuation that only runs once that re-thrown exception has faulted the
+ // shared task - so by the time this hook fires, self-healing has already happened. Using
+ // it to synchronize here (rather than a fixed Task.Delay) is exactly as deterministic as
+ // the timing it stands in for, with no arbitrary sleep to tune or risk racing under load.
+ TaskCompletionSource observedTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ store.OnInitializationFaultObserved = ex => observedTcs.TrySetResult(ex);
+
+ using CancellationTokenSource cts1 = new();
+ using CancellationTokenSource cts2 = new();
+ Task upload1 = store.UploadAsync("payload", cts1.Token);
+ Task upload2 = store.UploadAsync("payload", cts2.Token);
+
+ // Act: every caller waiting on the shared initialization abandons it (cancels its own
+ // token) before that shared initialization itself later fails. No caller remains to
+ // observe the failure directly, so only the initialization task's own completion path
+ // can self-heal the cache.
+ cts1.Cancel();
+ cts2.Cancel();
+ Func awaitUpload1 = () => upload1;
+ Func awaitUpload2 = () => upload2;
+ await awaitUpload1.Should().ThrowAsync();
+ await awaitUpload2.Should().ThrowAsync();
+
+ initTcs.SetException(new RequestFailedException(503, "Service unavailable"));
+
+ // Wait deterministically for the shared initialization task's own fault-observing
+ // continuation to run (and, with it, self-heal the cache) instead of a fixed delay: a
+ // generous timeout guards against the test hanging forever (instead of failing with a
+ // clear message) if that continuation never runs at all - which would itself indicate a
+ // regression in self-healing.
+ Task completedTask = await Task.WhenAny(observedTcs.Task, Task.Delay(TimeSpan.FromSeconds(5)));
+ completedTask.Should().BeSameAs(observedTcs.Task, "the shared initializer's fault-observing continuation (and self-heal) should have run within the timeout");
+
+ // Assert: a brand-new upload gets a fresh initialization attempt instead of reusing the
+ // now-stale failed one.
+ string token = await store.UploadAsync("payload", CancellationToken.None);
+ token.Should().NotBeNullOrEmpty();
+ containerClientMock.Verify(
+ c => c.CreateIfNotExistsAsync(
+ It.IsAny(),
+ It.IsAny>(),
+ It.IsAny(),
+ It.IsAny()),
+ Times.Exactly(2));
+ }
+
+ [Fact]
+ public void OnInitializationFaultObserved_DefaultsToNonNullNoOpAndRejectsNull()
+ {
+ // Arrange: construct a store exactly as production code does, without touching the hook.
+ BlobPayloadStore store = new(new LargePayloadStorageOptions(), CreateContainerClientMock().Object);
+
+ // Assert: the hook is never null - neither by default nor after explicitly assigning
+ // null - and invoking the untouched default does nothing (it's a no-op), not throw.
+ //
+ // This is what actually proves the production-default path is safe, independent of any
+ // test that overrides the hook. UploadAsync_AllWaitersCancelThenInitializerFaults_ExceptionIsObserved
+ // (below) proves the fault-observing continuation runs and reads Task.Exception when a
+ // custom hook is installed, but that alone can't distinguish "the continuation
+ // unconditionally reads Task.Exception" from "it only does so because a hook happens to
+ // be set" - which is exactly how the historical
+ // `this.OnInitializationFaultObserved?.Invoke(t.Exception!)` regression passed every
+ // existing test while silently never observing faults in production, where the hook was
+ // null by default. This test closes that gap directly: because the hook can never be
+ // null - by construction of this property, not by convention - the continuation's direct,
+ // unconditional invocation of it (see ObserveFaultWithoutAwaiting) always evaluates
+ // Task.Exception, whether or not any test has overridden the hook.
+ store.OnInitializationFaultObserved.Should().NotBeNull();
+
+ Action invokeDefault = () => store.OnInitializationFaultObserved(new InvalidOperationException("boom"));
+ invokeDefault.Should().NotThrow();
+
+ store.OnInitializationFaultObserved = null!;
+ store.OnInitializationFaultObserved.Should().NotBeNull();
+ }
+
+ [Fact]
+ public async Task UploadAsync_AllWaitersCancelThenInitializerFaults_ExceptionIsObserved()
+ {
+ // Arrange: control exactly when the shared initialization completes/fails, and use the
+ // internal OnInitializationFaultObserved test hook to deterministically prove that the
+ // fault-observing continuation attached in PublishNewInitializer actually ran and read
+ // the shared task's Exception - rather than relying on TaskScheduler.UnobservedTaskException
+ // plus a forced GC, which can't reliably distinguish "the fix ran" from "the CLR just
+ // hasn't collected the task yet" (e.g. because the async state machine still roots it).
+ // The continuation is attached once, up front, when the initializer is published, so this
+ // applies uniformly regardless of which cancellation code path any given caller takes.
+ //
+ // Overriding the hook here with a custom delegate exercises the exact same statement -
+ // "this.OnInitializationFaultObserved(t.Exception!)", no null-conditional - that runs
+ // against the untouched, no-op default in production (see
+ // OnInitializationFaultObserved_DefaultsToNonNullNoOpAndRejectsNull, which proves that
+ // default can never be null): the only difference is which Action instance
+ // gets invoked, not whether it does.
+ TaskCompletionSource> initTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ Mock containerClientMock = CreateContainerClientMock();
+ containerClientMock
+ .Setup(c => c.CreateIfNotExistsAsync(
+ It.IsAny(),
+ It.IsAny>(),
+ It.IsAny(),
+ It.IsAny()))
+ .Returns(initTcs.Task);
+
+ BlobPayloadStore store = new(new LargePayloadStorageOptions(), containerClientMock.Object);
+
+ int observedCount = 0;
+ TaskCompletionSource observedTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ store.OnInitializationFaultObserved = ex =>
+ {
+ Interlocked.Increment(ref observedCount);
+
+ // TrySetResult (rather than blindly SetResult) tolerates the hook firing more than
+ // once without throwing here; the exactly-once assertion below is what actually
+ // proves it fired exactly once.
+ observedTcs.TrySetResult(ex);
+ };
+
+ using CancellationTokenSource cts1 = new();
+ using CancellationTokenSource cts2 = new();
+
+ // Act: every caller of the still-pending shared initialization cancels its own token
+ // before that shared initialization later faults, so no caller is left waiting on it
+ // when the failure occurs.
+ Task upload1 = store.UploadAsync("payload", cts1.Token);
+ Task upload2 = store.UploadAsync("payload", cts2.Token);
+
+ cts1.Cancel();
+ cts2.Cancel();
+ Func awaitUpload1 = () => upload1;
+ Func awaitUpload2 = () => upload2;
+ await awaitUpload1.Should().ThrowAsync();
+ await awaitUpload2.Should().ThrowAsync();
+
+ RequestFailedException expectedException = new(503, "Service unavailable");
+ initTcs.SetException(expectedException);
+
+ // Wait deterministically for the fault-observing continuation to run instead of polling:
+ // it's attached with TaskContinuationOptions.ExecuteSynchronously, but SetException above
+ // may itself run continuations asynchronously depending on scheduling, so await the TCS
+ // it completes rather than assuming it already ran. A generous timeout guards against the
+ // test hanging forever (instead of failing with a clear message) if the continuation never
+ // runs at all - which would itself indicate a regression in the fix under test.
+ Task completedTask = await Task.WhenAny(observedTcs.Task, Task.Delay(TimeSpan.FromSeconds(5)));
+ completedTask.Should().BeSameAs(observedTcs.Task, "the fault-observing continuation should have run within the timeout");
+ Exception observedException = await observedTcs.Task;
+
+ containerClientMock.Verify(
+ c => c.CreateIfNotExistsAsync(
+ It.IsAny(),
+ It.IsAny>(),
+ It.IsAny(),
+ It.IsAny()),
+ Times.Once);
+
+ // Assert: the shared initializer's fault was observed exactly once, proving the
+ // continuation ran and read Task.Exception - not merely that the runtime hasn't yet
+ // reported it as unobserved. Task.Exception wraps the fault in an AggregateException,
+ // so unwrap it to confirm it's the exact exception the initializer faulted with.
+ observedCount.Should().Be(1);
+ observedException.Should().BeOfType();
+ ((AggregateException)observedException).InnerException.Should().BeSameAs(expectedException);
+ }
+
+ [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();
+ 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);
+ await Task.Delay(20);
+
+ // 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
+ .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;