From 0584899a8f031b952dbbffe1d57865b1306bb5e7 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 15:10:35 -0700 Subject: [PATCH 1/9] Cache Azure Blob container initialization to avoid per-upload CreateIfNotExistsAsync BlobPayloadStore.UploadAsync previously called CreateIfNotExistsAsync on every upload. This adds a single-flight cached initialization task so the container is created at most once per store instance, while: - remaining concurrency-safe for first use (concurrent callers share the same in-flight initialization task) - keeping initialization failures retriable (a faulted/canceled attempt is not cached, so the next caller retries) - honoring each caller's own CancellationToken without canceling the shared initialization for other callers - recovering automatically if the container is deleted after initialization (detected via BlobErrorCode.ContainerNotFound on write, which resets the cache so the next upload recreates the container) No public API changes. Added focused unit tests covering single-flight caching, retry-after-failure, cancellation isolation, container-deletion recovery, and unrelated-error propagation. Fixes #771 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6fcfc0e9-3ac1-476b-adac-9ad37d7d1b12 --- Microsoft.DurableTask.sln | 18 ++ .../PayloadStore/BlobPayloadStore.cs | 150 ++++++++++-- .../AzureBlobPayloads.Tests.csproj | 16 ++ .../BlobPayloadStoreTests.cs | 228 ++++++++++++++++++ .../AzureBlobPayloads.Tests/Usings.cs | 6 + 5 files changed, 398 insertions(+), 20 deletions(-) create mode 100644 test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj create mode 100644 test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs create mode 100644 test/Extensions/AzureBlobPayloads.Tests/Usings.cs 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..70ee89f5 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -26,6 +26,11 @@ public sealed class BlobPayloadStore : PayloadStore readonly BlobContainerClient containerClient; readonly LargePayloadStorageOptions options; + // Caches the single in-flight (or completed) container-initialization task so that + // concurrent/subsequent uploads don't each issue their own CreateIfNotExistsAsync request. + // Null means "not yet attempted" or "needs to be retried"; see EnsureContainerExistsAsync. + Task? containerInitializationTask; + /// /// Initializes a new instance of the class. /// @@ -67,6 +72,18 @@ 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)); + } + /// public override async Task UploadAsync(string payLoad, CancellationToken cancellationToken) { @@ -76,33 +93,45 @@ 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); + // 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. + await this.EnsureContainerExistsAsync(cancellationToken).ConfigureAwait(false); - if (this.options.CompressionEnabled) + try { - BlobOpenWriteOptions writeOptions = new() + if (this.options.CompressionEnabled) { - 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); + 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); + // 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); + } } - else + catch (RequestFailedException ex) when (ex.ErrorCode == BlobErrorCode.ContainerNotFound) { - 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); + // The container existed when we last verified/created it but has since been deleted + // (e.g. by an operator). Clear the cached initialization state so the next upload + // attempts to recreate the container, keeping deliberate deletion recoverable. + Volatile.Write(ref this.containerInitializationTask, null); + throw; } return EncodeToken(this.containerClient.Name, blobName); @@ -156,6 +185,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 +253,56 @@ 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. The result is cached for the lifetime of this + /// instance once it completes successfully; a failed attempt is not cached, so the next + /// caller retries. + /// + async Task EnsureContainerExistsAsync(CancellationToken cancellationToken) + { + Task initializationTask = Volatile.Read(ref this.containerInitializationTask) + ?? this.BeginContainerInitialization(); + + try + { + await WaitForInitializationAsync(initializationTask, cancellationToken).ConfigureAwait(false); + } + finally + { + if (initializationTask.IsFaulted || initializationTask.IsCanceled) + { + // Don't cache a failed attempt (e.g. a transient/throttled storage error): allow + // the next caller to retry initialization instead of failing forever. Use + // CompareExchange so we don't clobber a newer task set by a racing caller. + _ = Interlocked.CompareExchange(ref this.containerInitializationTask, null, initializationTask); + } + } + } + + /// + /// Starts container initialization if it hasn't already started, in a single-flight manner: + /// only the first caller's task is stored and returned to all callers, including ones racing + /// concurrently on other threads. + /// + Task BeginContainerInitialization() + { + Task newInitializationTask = this.CreateContainerIfNotExistsAsync(); + return Interlocked.CompareExchange(ref this.containerInitializationTask, newInitializationTask, null) + ?? newInitializationTask; + } + + /// + /// 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). + /// + async Task CreateContainerIfNotExistsAsync() + { + await this.containerClient.CreateIfNotExistsAsync(PublicAccessType.None, cancellationToken: CancellationToken.None) + .ConfigureAwait(false); + } } 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..51457e36 --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs @@ -0,0 +1,228 @@ +// 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 + 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: fire many concurrent uploads against a fresh (uninitialized) store. + Task[] uploads = Enumerable.Range(0, 10) + .Select(_ => 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(); + 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(new MemoryStream()); + 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 container-not-found failure reset the cache, so the second upload + // recreated the container instead of assuming it still existed. + secondToken.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 + .SetupSequence(b => b.OpenWriteAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new RequestFailedException(500, "Internal server error")) + .ReturnsAsync(new MemoryStream()); + 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); + } + + 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; From 46d028a67bde2786910b28e621592fb59ccb083c Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 16:26:18 -0700 Subject: [PATCH 2/9] Fix concurrency flaws in container initialization single-flight design Address 3 medium-severity concurrency issues found in code review of PR #785: 1. True single-flight: PublishNewInitializer now atomically publishes an unstarted Lazy gate via CompareExchange before any real work starts, so racing first-time callers can never each independently trigger their own CreateIfNotExistsAsync call. Lazy with LazyThreadSafetyMode.ExecutionAndPublication guarantees the factory (the real SDK call) runs exactly once even under concurrent Value access. 2. Self-healing cache: CreateContainerIfNotExistsAsync now clears the cached initializer from its own completion path (a catch block keyed off the Lazy instance itself) whenever it fails, independent of whether any caller is still around to observe the failure. Previously, cleanup only happened in each waiting caller's own code path, so if every waiter cancelled before the shared initialization later faulted, the stale failure was cached forever. 3. Generation-aware recovery: the ContainerNotFound catch in UploadAsync now does a CompareExchange against the specific initializer instance the upload used, instead of an unconditional write. This ensures a stale deletion-recovery attempt can never clobber a newer initializer already published by a faster-recovering concurrent upload. Also replaces the sequential LINQ 'concurrency' test (which never actually raced, since Task.WhenAll doesn't force concurrent entry) with a Barrier-gated Task.Run-based test that forces genuinely concurrent workers and verifies exactly one create call. Adds two new deterministic regression tests: all-waiters-cancel-then-init-fails-then-retry, and overlapping stale ContainerNotFound failures vs. a newer initializer. No public API changes. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6fcfc0e9-3ac1-476b-adac-9ad37d7d1b12 --- .../PayloadStore/BlobPayloadStore.cs | 95 +++++++---- .../BlobPayloadStoreTests.cs | 158 +++++++++++++++++- 2 files changed, 213 insertions(+), 40 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs index 70ee89f5..5e903279 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -26,10 +26,14 @@ public sealed class BlobPayloadStore : PayloadStore readonly BlobContainerClient containerClient; readonly LargePayloadStorageOptions options; - // Caches the single in-flight (or completed) container-initialization task so that + // 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"; see EnsureContainerExistsAsync. - Task? containerInitializationTask; + // 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; /// /// Initializes a new instance of the class. @@ -95,7 +99,9 @@ public override async Task UploadAsync(string payLoad, CancellationToken // 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. - await this.EnsureContainerExistsAsync(cancellationToken).ConfigureAwait(false); + // Keep the specific initializer instance this upload used so the ContainerNotFound + // recovery below can invalidate it precisely (see the catch block). + Lazy containerInitializer = await this.EnsureContainerExistsAsync(cancellationToken).ConfigureAwait(false); try { @@ -128,9 +134,12 @@ public override async Task UploadAsync(string payLoad, CancellationToken catch (RequestFailedException ex) when (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 initialization state so the next upload - // attempts to recreate the container, keeping deliberate deletion recoverable. - Volatile.Write(ref this.containerInitializationTask, null); + // (e.g. by an operator). Clear the cached initializer so the next upload attempts to + // recreate the container, keeping deliberate deletion recoverable. CompareExchange + // against the specific initializer this upload used ensures a stale failure can + // never clobber a newer initializer already published by another, faster-recovering + // concurrent upload that detected and recovered from the same deletion. + _ = Interlocked.CompareExchange(ref this.containerInitializer, null, containerInitializer); throw; } @@ -256,41 +265,38 @@ static async Task ReadToEndAsync(StreamReader reader, CancellationToken /// /// Ensures the container exists, issuing at most one CreateIfNotExistsAsync request - /// across all concurrent/subsequent callers. The result is cached for the lifetime of this - /// instance once it completes successfully; a failed attempt is not cached, so the next - /// caller retries. + /// 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) + async Task> EnsureContainerExistsAsync(CancellationToken cancellationToken) { - Task initializationTask = Volatile.Read(ref this.containerInitializationTask) - ?? this.BeginContainerInitialization(); + Lazy initializer = Volatile.Read(ref this.containerInitializer) + ?? this.PublishNewInitializer(); - try - { - await WaitForInitializationAsync(initializationTask, cancellationToken).ConfigureAwait(false); - } - finally - { - if (initializationTask.IsFaulted || initializationTask.IsCanceled) - { - // Don't cache a failed attempt (e.g. a transient/throttled storage error): allow - // the next caller to retry initialization instead of failing forever. Use - // CompareExchange so we don't clobber a newer task set by a racing caller. - _ = Interlocked.CompareExchange(ref this.containerInitializationTask, null, initializationTask); - } - } + await WaitForInitializationAsync(initializer.Value, cancellationToken).ConfigureAwait(false); + return initializer; } /// - /// Starts container initialization if it hasn't already started, in a single-flight manner: - /// only the first caller's task is stored and returned to all callers, including ones racing - /// concurrently on other threads. + /// 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. /// - Task BeginContainerInitialization() + Lazy PublishNewInitializer() { - Task newInitializationTask = this.CreateContainerIfNotExistsAsync(); - return Interlocked.CompareExchange(ref this.containerInitializationTask, newInitializationTask, null) - ?? newInitializationTask; + Lazy? initializer = null; + initializer = new Lazy( + () => this.CreateContainerIfNotExistsAsync(initializer!), + LazyThreadSafetyMode.ExecutionAndPublication); + + return Interlocked.CompareExchange(ref this.containerInitializer, initializer, null) ?? initializer; } /// @@ -299,10 +305,25 @@ Task BeginContainerInitialization() /// 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() + async Task CreateContainerIfNotExistsAsync(Lazy self) { - await this.containerClient.CreateIfNotExistsAsync(PublicAccessType.None, cancellationToken: CancellationToken.None) - .ConfigureAwait(false); + 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/BlobPayloadStoreTests.cs b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs index 51457e36..79b4484d 100644 --- a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs +++ b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs @@ -18,6 +18,7 @@ public class BlobPayloadStoreTests public async Task UploadAsync_ConcurrentCallers_OnlyCreatesContainerOnce() { // Arrange + const int WorkerCount = 16; int createCalls = 0; Mock containerClientMock = CreateContainerClientMock(); containerClientMock @@ -35,9 +36,21 @@ public async Task UploadAsync_ConcurrentCallers_OnlyCreatesContainerOnce() BlobPayloadStore store = new(new LargePayloadStorageOptions(), containerClientMock.Object); - // Act: fire many concurrent uploads against a fresh (uninitialized) store. - Task[] uploads = Enumerable.Range(0, 10) - .Select(_ => store.UploadAsync("payload", CancellationToken.None)) + // 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); @@ -209,6 +222,145 @@ public async Task UploadAsync_UnrelatedStorageError_PropagatesWithoutInvalidatin 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); + + 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")); + + // Give the shared initialization task's own continuation a chance to run and self-heal + // the cache, even though no caller is left waiting on it. + await Task.Delay(100); + + // 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 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(); From 79b1d063779f28aaf721b64b55be6eb6ec98cb63 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 16:55:48 -0700 Subject: [PATCH 3/9] Observe shared init task fault in netstandard2.0 cancellation path Address a Medium issue from Terra's final re-review of PR #785: in the NETSTANDARD2_0-only branch of WaitForInitializationAsync, when a caller's own cancellation token fires first, it throws OperationCanceledException without ever awaiting the shared initialization task. If every waiter abandons the task this way and it later faults, nobody observes its exception, which the runtime reports via TaskScheduler.UnobservedTaskException on finalization. Fix: attach a fire-and-forget OnlyOnFaulted continuation that touches Task.Exception when a caller abandons the shared task due to its own cancellation, so the fault is always observed regardless of whether any caller stays around to await it. This doesn't block or delay the caller's own cancellation, and is independent of the existing cache self-healing logic in CreateContainerIfNotExistsAsync. Since the test project only targets a single runnable framework (not netstandard2.0), the ifdef'd branch itself can't be directly exercised by xunit. Adds a framework-neutral test that proves the underlying fault-observation pattern (OnlyOnFaulted continuation touching .Exception) prevents TaskScheduler.UnobservedTaskException, using a GC/finalization pass to verify no unobserved exception is reported. Verified via dotnet build across all 4 target frameworks (netstandard2.0, net6.0, net8.0, net10.0): 0 errors, no new warnings. All 8 unit tests pass, including the new test run 8x for stability. No public API changes; cancellation and cache self-healing semantics are unaffected. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6fcfc0e9-3ac1-476b-adac-9ad37d7d1b12 --- .../PayloadStore/BlobPayloadStore.cs | 13 +++++ .../BlobPayloadStoreTests.cs | 57 +++++++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs index 5e903279..9ec33c66 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -213,6 +213,19 @@ static async Task WaitForInitializationAsync(Task initializationTask, Cancellati Task completed = await Task.WhenAny(initializationTask, cancellationTcs.Task).ConfigureAwait(false); if (completed == cancellationTcs.Task) { + // This caller is abandoning the shared initializationTask without ever awaiting + // it below. If every other caller does the same and the task later faults, no + // one would ever observe its exception, which the runtime reports (on + // finalization) via TaskScheduler.UnobservedTaskException. Attach a + // fire-and-forget continuation that touches the exception on fault so it's + // always observed, without awaiting/blocking on it here - that would delay or + // change this caller's own cancellation semantics - and without affecting the + // separate cache self-healing logic in CreateContainerIfNotExistsAsync. + _ = initializationTask.ContinueWith( + static t => _ = t.Exception, + CancellationToken.None, + TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); cancellationToken.ThrowIfCancellationRequested(); } } diff --git a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs index 79b4484d..0e0b2f4e 100644 --- a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs +++ b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs @@ -361,6 +361,63 @@ public async Task UploadAsync_StaleContainerNotFoundFailure_DoesNotOverwriteNewe createCalls.Should().Be(2); } + /// + /// Proves the fault-observation pattern relied on by the netstandard2.0-only cancellation + /// branch of BlobPayloadStore.WaitForInitializationAsync: when a caller abandons the + /// shared initialization task because its own cancellation token fires first, it attaches a + /// fire-and-forget continuation that + /// touches so the shared task's eventual fault is always + /// "observed" - even if every caller abandons it this way - and never surfaces via + /// when the task is later finalized. + /// This test cannot exercise the netstandard2.0-only source directly (this test project + /// targets a single runnable framework, not netstandard2.0), so it instead verifies the + /// underlying pattern in isolation. + /// + [Fact] + public async Task FaultObservingContinuation_PreventsUnobservedTaskException() + { + // Arrange + TaskCompletionSource tcs = new(TaskCreationOptions.RunContinuationsAsynchronously); + bool unobserved = false; + void OnUnobserved(object? sender, UnobservedTaskExceptionEventArgs e) + { + unobserved = true; + e.SetObserved(); + } + + TaskScheduler.UnobservedTaskException += OnUnobserved; + try + { + // Act: apply the same "abandon but still observe on fault" pattern used by + // WaitForInitializationAsync's netstandard2.0 cancellation branch, then let the + // shared task fault after it has already been abandoned. + Task sharedTask = tcs.Task; + _ = sharedTask.ContinueWith( + static t => _ = t.Exception, + CancellationToken.None, + TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + tcs.SetException(new InvalidOperationException("boom")); + + // Let the fault-observing continuation run. + await Task.Delay(50); + + // Drop the last strong reference and force finalization, which is when the runtime + // reports any still-unobserved task exceptions. + sharedTask = null!; + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + } + finally + { + TaskScheduler.UnobservedTaskException -= OnUnobserved; + } + + // Assert: the fault was observed, so it was never reported as unobserved. + unobserved.Should().BeFalse(); + } + static Mock CreateContainerClientMock() { Mock containerClientMock = new(); From 2be3fb12719b521d55c2522b6741b99585aed895 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 17:08:31 -0700 Subject: [PATCH 4/9] Consolidate fault observation into single-attach point in PublishNewInitializer Terra's re-review found the previous fault-observation fix only covered the NETSTANDARD2_0 cancellation branch of WaitForInitializationAsync; on modern TFMs (WaitAsync-based cancellation), if every caller cancels before the shared initializer later faults, the fault remained unobserved (reproduced on net10.0). Move fault observation out of WaitForInitializationAsync entirely and attach it exactly once, in PublishNewInitializer, by whichever caller wins the CompareExchange publish race. This decouples observation from any particular caller's cancellation code path, fixing both the NETSTANDARD2_0 and WaitAsync-based branches with a single mechanism, while preserving the publish-before-work-starts single-flight invariant (the CompareExchange still happens before .Value is accessed). Revert the now-redundant per-caller ContinueWith block in the NETSTANDARD2_0 branch back to a plain ThrowIfCancellationRequested. Add a production regression test exercising the WaitAsync cancellation path (net10.0): all callers cancel, then the shared initializer faults, and TaskScheduler.UnobservedTaskException is asserted to never fire. Verified the test fails without the fix (temporarily disabled the observer call) and passes with it, confirming it is not a false positive. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6fcfc0e9-3ac1-476b-adac-9ad37d7d1b12 --- .../PayloadStore/BlobPayloadStore.cs | 47 +++++++---- .../BlobPayloadStoreTests.cs | 80 +++++++++++++++++++ 2 files changed, 112 insertions(+), 15 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs index 9ec33c66..cbd10ec2 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -213,19 +213,6 @@ static async Task WaitForInitializationAsync(Task initializationTask, Cancellati Task completed = await Task.WhenAny(initializationTask, cancellationTcs.Task).ConfigureAwait(false); if (completed == cancellationTcs.Task) { - // This caller is abandoning the shared initializationTask without ever awaiting - // it below. If every other caller does the same and the task later faults, no - // one would ever observe its exception, which the runtime reports (on - // finalization) via TaskScheduler.UnobservedTaskException. Attach a - // fire-and-forget continuation that touches the exception on fault so it's - // always observed, without awaiting/blocking on it here - that would delay or - // change this caller's own cancellation semantics - and without affecting the - // separate cache self-healing logic in CreateContainerIfNotExistsAsync. - _ = initializationTask.ContinueWith( - static t => _ = t.Exception, - CancellationToken.None, - TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, - TaskScheduler.Default); cancellationToken.ThrowIfCancellationRequested(); } } @@ -276,6 +263,24 @@ static async Task ReadToEndAsync(StreamReader reader, CancellationToken return (rest.Substring(0, sep), rest.Substring(sep + 1)); } + /// + /// Attaches a fire-and-forget continuation that touches 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. + /// + static void ObserveFaultWithoutAwaiting(Task task) + { + _ = task.ContinueWith( + static t => _ = t.Exception, + CancellationToken.None, + TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + } + /// /// Ensures the container exists, issuing at most one CreateIfNotExistsAsync request /// across all concurrent/subsequent callers, and returns the specific initializer gate that @@ -300,7 +305,13 @@ async Task> EnsureContainerExistsAsync(CancellationToken cancellation /// 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. + /// 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() { @@ -309,7 +320,13 @@ Lazy PublishNewInitializer() () => this.CreateContainerIfNotExistsAsync(initializer!), LazyThreadSafetyMode.ExecutionAndPublication); - return Interlocked.CompareExchange(ref this.containerInitializer, initializer, null) ?? initializer; + Lazy published = Interlocked.CompareExchange(ref this.containerInitializer, initializer, null) ?? initializer; + if (ReferenceEquals(published, initializer)) + { + ObserveFaultWithoutAwaiting(published.Value); + } + + return published; } /// diff --git a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs index 0e0b2f4e..a8fa03d2 100644 --- a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs +++ b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs @@ -276,6 +276,86 @@ public async Task UploadAsync_AllWaitersCancelBeforeInitializationFails_NextUplo Times.Exactly(2)); } + [Fact] + public async Task UploadAsync_AllWaitersCancelThenInitializerFaults_ExceptionIsObserved() + { + // Arrange: control exactly when the shared initialization completes/fails, and observe + // whether the runtime ever reports its eventual exception as unobserved. This exercises + // the WaitAsync-based cancellation path used by WaitForInitializationAsync on this + // (non-netstandard2.0) target framework - the fault-observing continuation is attached + // once, up front, when the initializer is published (see PublishNewInitializer), so it + // applies uniformly regardless of which cancellation code path any given caller takes. + 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); + + bool unobserved = false; + void OnUnobserved(object? sender, UnobservedTaskExceptionEventArgs e) + { + unobserved = true; + e.SetObserved(); + } + + TaskScheduler.UnobservedTaskException += OnUnobserved; + try + { + 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(); + + initTcs.SetException(new RequestFailedException(503, "Service unavailable")); + + // Give the fault-observing continuation attached when the initializer was published + // a chance to run. + await Task.Delay(100); + + containerClientMock.Verify( + c => c.CreateIfNotExistsAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Once); + + // Drop every reference to the store (and thus, transitively, to the shared + // initializer task) and force a full GC + finalization pass - the point at which the + // runtime reports any still-unobserved task exceptions. + upload1 = null; + upload2 = null; + store = null; + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + } + finally + { + TaskScheduler.UnobservedTaskException -= OnUnobserved; + } + + // Assert: the shared initializer's fault was observed, so it was never reported. + unobserved.Should().BeFalse(); + } + [Fact] public async Task UploadAsync_StaleContainerNotFoundFailure_DoesNotOverwriteNewerInitializer() { From 41025473e639f266e2d7460af881b4479e202ec5 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 17:22:47 -0700 Subject: [PATCH 5/9] Replace GC-based fault-observation test with deterministic hook Terra's final review flagged that the cancellation/fault regression test relied on TaskScheduler.UnobservedTaskException plus a forced GC while still holding references reachable from the async state machine, making pass/fail dependent on GC/finalization timing, debugger attachment, and JIT optimizations rather than purely on whether the fix is present. Add an internal Action? OnInitializationFaultObserved property on BlobPayloadStore, invoked by ObserveFaultWithoutAwaiting (now an instance method) immediately after it reads the faulted task's Exception. Defaults to null in production (no-op, zero overhead); each test constructs its own store instance so there's no shared state to reset. Rewrite UploadAsync_AllWaitersCancelThenInitializerFaults_ExceptionIsObserved to set this hook to a thread-safe (Interlocked/Volatile) counter + captured exception instead of subscribing to UnobservedTaskException and forcing a GC, and assert the continuation fired exactly once with the expected fault. Verified the test correctly fails when the hook wiring is temporarily removed, and passes across 8 repeat runs once restored. No public API change (the new member is internal); no production behavior change (the hook is a no-op unless a test sets it). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6fcfc0e9-3ac1-476b-adac-9ad37d7d1b12 --- .../PayloadStore/BlobPayloadStore.cs | 55 ++++++--- .../BlobPayloadStoreTests.cs | 112 +++++++++--------- 2 files changed, 90 insertions(+), 77 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs index cbd10ec2..98adf05a 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -88,6 +88,22 @@ internal BlobPayloadStore(LargePayloadStorageOptions options, BlobContainerClien this.containerClient = containerClient ?? throw new ArgumentNullException(nameof(containerClient)); } + /// + /// Gets or sets a test-only hook, invoked at most once per initialization attempt, + /// 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". Left (the default) + /// in production, where invoking it is a no-op; 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; set; } + /// public override async Task UploadAsync(string payLoad, CancellationToken cancellationToken) { @@ -263,24 +279,6 @@ static async Task ReadToEndAsync(StreamReader reader, CancellationToken return (rest.Substring(0, sep), rest.Substring(sep + 1)); } - /// - /// Attaches a fire-and-forget continuation that touches 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. - /// - static void ObserveFaultWithoutAwaiting(Task task) - { - _ = task.ContinueWith( - static t => _ = t.Exception, - CancellationToken.None, - TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, - TaskScheduler.Default); - } - /// /// Ensures the container exists, issuing at most one CreateIfNotExistsAsync request /// across all concurrent/subsequent callers, and returns the specific initializer gate that @@ -323,12 +321,31 @@ Lazy PublishNewInitializer() Lazy published = Interlocked.CompareExchange(ref this.containerInitializer, initializer, null) ?? initializer; if (ReferenceEquals(published, initializer)) { - ObserveFaultWithoutAwaiting(published.Value); + 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. + /// + void ObserveFaultWithoutAwaiting(Task task) + { + _ = task.ContinueWith( + t => this.OnInitializationFaultObserved?.Invoke(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 ; diff --git a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs index a8fa03d2..25083ad4 100644 --- a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs +++ b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs @@ -279,11 +279,13 @@ public async Task UploadAsync_AllWaitersCancelBeforeInitializationFails_NextUplo [Fact] public async Task UploadAsync_AllWaitersCancelThenInitializerFaults_ExceptionIsObserved() { - // Arrange: control exactly when the shared initialization completes/fails, and observe - // whether the runtime ever reports its eventual exception as unobserved. This exercises - // the WaitAsync-based cancellation path used by WaitForInitializationAsync on this - // (non-netstandard2.0) target framework - the fault-observing continuation is attached - // once, up front, when the initializer is published (see PublishNewInitializer), so it + // 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. TaskCompletionSource> initTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); Mock containerClientMock = CreateContainerClientMock(); @@ -295,65 +297,59 @@ public async Task UploadAsync_AllWaitersCancelThenInitializerFaults_ExceptionIsO It.IsAny())) .Returns(initTcs.Task); - BlobPayloadStore? store = new(new LargePayloadStorageOptions(), containerClientMock.Object); + BlobPayloadStore store = new(new LargePayloadStorageOptions(), containerClientMock.Object); - bool unobserved = false; - void OnUnobserved(object? sender, UnobservedTaskExceptionEventArgs e) + int observedCount = 0; + Exception? observedException = null; + store.OnInitializationFaultObserved = ex => { - unobserved = true; - e.SetObserved(); - } + Interlocked.Increment(ref observedCount); + Volatile.Write(ref observedException, ex); + }; - TaskScheduler.UnobservedTaskException += OnUnobserved; - try - { - 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(); - - initTcs.SetException(new RequestFailedException(503, "Service unavailable")); - - // Give the fault-observing continuation attached when the initializer was published - // a chance to run. - await Task.Delay(100); - - containerClientMock.Verify( - c => c.CreateIfNotExistsAsync( - It.IsAny(), - It.IsAny>(), - It.IsAny(), - It.IsAny()), - Times.Once); - - // Drop every reference to the store (and thus, transitively, to the shared - // initializer task) and force a full GC + finalization pass - the point at which the - // runtime reports any still-unobserved task exceptions. - upload1 = null; - upload2 = null; - store = null; - GC.Collect(); - GC.WaitForPendingFinalizers(); - GC.Collect(); - } - finally + 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); + + // Give the fault-observing continuation attached when the initializer was published a + // chance to run (it's attached with TaskContinuationOptions.ExecuteSynchronously, but + // SetException above may itself run continuations asynchronously depending on scheduling, + // so poll briefly instead of assuming it already ran). + for (int i = 0; i < 100 && Volatile.Read(ref observedCount) == 0; i++) { - TaskScheduler.UnobservedTaskException -= OnUnobserved; + await Task.Delay(10); } - // Assert: the shared initializer's fault was observed, so it was never reported. - unobserved.Should().BeFalse(); + 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] From 4529c666fcf0183870865a2cb1444f9daffe35cf Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 17:31:34 -0700 Subject: [PATCH 6/9] Fix unobserved fault when hook is null; remove stale test; TCS over polling Fixes a critical regression introduced by the previous commit: the fault continuation used his.OnInitializationFaultObserved?.Invoke(t.Exception!), which short-circuits and never evaluates .Exception when the hook is null (the production default). This meant the shared initializer's fault was no longer actually being observed in production whenever every caller cancelled before it faulted - reintroducing the original UnobservedTaskException risk this method exists to prevent. Fixed by unconditionally reading Task.Exception into a local first, then optionally invoking the hook with it. Also addresses two review nits: - Removed FaultObservingContinuation_PreventsUnobservedTaskException, a standalone test that never invoked BlobPayloadStore and described netstandard2.0-specific cancellation-path behavior that no longer exists (the continuation is now attached once in PublishNewInitializer regardless of TFM). The behavior it aimed to cover is already exercised end-to-end by UploadAsync_AllWaitersCancelThenInitializerFaults_ExceptionIsObserved. - Replaced that test's up-to-1s polling loop for the fault-observing hook with a TaskCompletionSource the hook completes, awaited via Task.WhenAny with a 5s timeout guard (fails with a clear assertion message instead of polling or hanging indefinitely). No public API change; no other production behavior change. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6fcfc0e9-3ac1-476b-adac-9ad37d7d1b12 --- .../PayloadStore/BlobPayloadStore.cs | 10 ++- .../BlobPayloadStoreTests.cs | 84 ++++--------------- 2 files changed, 25 insertions(+), 69 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs index 98adf05a..5d1dffd5 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -340,7 +340,15 @@ Lazy PublishNewInitializer() void ObserveFaultWithoutAwaiting(Task task) { _ = task.ContinueWith( - t => this.OnInitializationFaultObserved?.Invoke(t.Exception!), + t => + { + // Read Task.Exception unconditionally so the fault is always marked "observed", + // even in production where OnInitializationFaultObserved is null. Using + // "?.Invoke(t.Exception!)" here would short-circuit and never evaluate + // t.Exception when the hook is null, defeating the entire point of this method. + Exception exception = t.Exception!; + this.OnInitializationFaultObserved?.Invoke(exception); + }, CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); diff --git a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs index 25083ad4..0342d936 100644 --- a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs +++ b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs @@ -300,11 +300,15 @@ public async Task UploadAsync_AllWaitersCancelThenInitializerFaults_ExceptionIsO BlobPayloadStore store = new(new LargePayloadStorageOptions(), containerClientMock.Object); int observedCount = 0; - Exception? observedException = null; + TaskCompletionSource observedTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); store.OnInitializationFaultObserved = ex => { Interlocked.Increment(ref observedCount); - Volatile.Write(ref observedException, ex); + + // 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(); @@ -326,14 +330,15 @@ public async Task UploadAsync_AllWaitersCancelThenInitializerFaults_ExceptionIsO RequestFailedException expectedException = new(503, "Service unavailable"); initTcs.SetException(expectedException); - // Give the fault-observing continuation attached when the initializer was published a - // chance to run (it's attached with TaskContinuationOptions.ExecuteSynchronously, but - // SetException above may itself run continuations asynchronously depending on scheduling, - // so poll briefly instead of assuming it already ran). - for (int i = 0; i < 100 && Volatile.Read(ref observedCount) == 0; i++) - { - await Task.Delay(10); - } + // 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( @@ -349,7 +354,7 @@ public async Task UploadAsync_AllWaitersCancelThenInitializerFaults_ExceptionIsO // 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); + ((AggregateException)observedException).InnerException.Should().BeSameAs(expectedException); } [Fact] @@ -437,63 +442,6 @@ public async Task UploadAsync_StaleContainerNotFoundFailure_DoesNotOverwriteNewe createCalls.Should().Be(2); } - /// - /// Proves the fault-observation pattern relied on by the netstandard2.0-only cancellation - /// branch of BlobPayloadStore.WaitForInitializationAsync: when a caller abandons the - /// shared initialization task because its own cancellation token fires first, it attaches a - /// fire-and-forget continuation that - /// touches so the shared task's eventual fault is always - /// "observed" - even if every caller abandons it this way - and never surfaces via - /// when the task is later finalized. - /// This test cannot exercise the netstandard2.0-only source directly (this test project - /// targets a single runnable framework, not netstandard2.0), so it instead verifies the - /// underlying pattern in isolation. - /// - [Fact] - public async Task FaultObservingContinuation_PreventsUnobservedTaskException() - { - // Arrange - TaskCompletionSource tcs = new(TaskCreationOptions.RunContinuationsAsynchronously); - bool unobserved = false; - void OnUnobserved(object? sender, UnobservedTaskExceptionEventArgs e) - { - unobserved = true; - e.SetObserved(); - } - - TaskScheduler.UnobservedTaskException += OnUnobserved; - try - { - // Act: apply the same "abandon but still observe on fault" pattern used by - // WaitForInitializationAsync's netstandard2.0 cancellation branch, then let the - // shared task fault after it has already been abandoned. - Task sharedTask = tcs.Task; - _ = sharedTask.ContinueWith( - static t => _ = t.Exception, - CancellationToken.None, - TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, - TaskScheduler.Default); - tcs.SetException(new InvalidOperationException("boom")); - - // Let the fault-observing continuation run. - await Task.Delay(50); - - // Drop the last strong reference and force finalization, which is when the runtime - // reports any still-unobserved task exceptions. - sharedTask = null!; - GC.Collect(); - GC.WaitForPendingFinalizers(); - GC.Collect(); - } - finally - { - TaskScheduler.UnobservedTaskException -= OnUnobserved; - } - - // Assert: the fault was observed, so it was never reported as unobserved. - unobserved.Should().BeFalse(); - } - static Mock CreateContainerClientMock() { Mock containerClientMock = new(); From 3513558039cdf43b7c6b3bd5ea6797eeb2077397 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 17:42:34 -0700 Subject: [PATCH 7/9] Make fault-observer hook non-nullable to eliminate short-circuit bug class by construction Redesign OnInitializationFaultObserved from a nullable auto-property to a non-nullable property backed by a private field that defaults to, and normalizes null assignments back to, a shared no-op delegate. This lets ObserveFaultWithoutAwaiting invoke the hook directly with no null-conditional operator, so the exact same statement executes whether or not a test has overridden the hook - closing a coverage gap where the previous nullable-hook design meant every existing test set a non-null hook, and so could never distinguish the fix (unconditional read of Task.Exception) from the historical short-circuiting bug (hook?.Invoke(t.Exception!), which never evaluates Task.Exception when the hook is null in production). Add OnInitializationFaultObserved_DefaultsToNonNullNoOpAndRejectsNull, which proves the untouched production default is never null and that assigning null resets it to the no-op rather than making it null - independent of any test that installs a custom hook. Verified this test fails under the old nullable/short-circuiting design and passes under the fix. No public API change (member remains internal); negligible production overhead (one extra no-op delegate invocation on the rare fault path). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6fcfc0e9-3ac1-476b-adac-9ad37d7d1b12 --- .../PayloadStore/BlobPayloadStore.cs | 65 ++++++++++++++----- .../BlobPayloadStoreTests.cs | 37 +++++++++++ 2 files changed, 85 insertions(+), 17 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs index 5d1dffd5..dc8b2abb 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -23,6 +23,15 @@ 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; @@ -35,6 +44,12 @@ public sealed class BlobPayloadStore : PayloadStore // 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. /// @@ -89,20 +104,41 @@ internal BlobPayloadStore(LargePayloadStorageOptions options, BlobContainerClien } /// - /// Gets or sets a test-only hook, invoked at most once per initialization attempt, - /// immediately after the fault-observing continuation attached in + /// 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". Left (the default) - /// in production, where invoking it is a no-op; each test uses its own - /// instance, so no reset between tests is needed. Callers of - /// this hook must not assume any particular thread. + /// "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; set; } + internal Action OnInitializationFaultObserved + { + get => this.onInitializationFaultObserved; + set => this.onInitializationFaultObserved = value ?? NoOpFaultObserver; + } /// public override async Task UploadAsync(string payLoad, CancellationToken cancellationToken) @@ -335,20 +371,15 @@ Lazy PublishNewInitializer() /// 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. + /// 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 => - { - // Read Task.Exception unconditionally so the fault is always marked "observed", - // even in production where OnInitializationFaultObserved is null. Using - // "?.Invoke(t.Exception!)" here would short-circuit and never evaluate - // t.Exception when the hook is null, defeating the entire point of this method. - Exception exception = t.Exception!; - this.OnInitializationFaultObserved?.Invoke(exception); - }, + t => this.OnInitializationFaultObserved(t.Exception!), CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); diff --git a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs index 0342d936..b2e2b57c 100644 --- a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs +++ b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs @@ -276,6 +276,36 @@ public async Task UploadAsync_AllWaitersCancelBeforeInitializationFails_NextUplo 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() { @@ -287,6 +317,13 @@ public async Task UploadAsync_AllWaitersCancelThenInitializerFaults_ExceptionIsO // 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 From a7a6d37a651ad3d6d0c4a20fc15dfaa12cac9716 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 17:52:26 -0700 Subject: [PATCH 8/9] Test: replace fixed delay with fault-observer coordination in self-heal test UploadAsync_AllWaitersCancelBeforeInitializationFails_NextUploadRetriesWithFreshAttempt used 'await Task.Delay(100)' to give the shared initializer's self-healing completion path a chance to run before retrying, with no guarantee the cache-clearing continuation had actually completed in that window. Replace it with the existing OnInitializationFaultObserved TaskCompletionSource coordination pattern: the fault-observing continuation only runs after CreateContainerIfNotExistsAsync has already cleared the cache in its catch block (strictly before re-throwing), so awaiting that hook - with a timeout diagnostic guard - deterministically waits for self-healing to complete without any arbitrary sleep to tune or risk racing under load. No production behavior change. Verified 9/9 full suite and 8x repeat of the fixed test for stability; all 4 TFMs build clean. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6fcfc0e9-3ac1-476b-adac-9ad37d7d1b12 --- .../BlobPayloadStoreTests.cs | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs index b2e2b57c..16dc2fdc 100644 --- a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs +++ b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs @@ -241,6 +241,15 @@ public async Task UploadAsync_AllWaitersCancelBeforeInitializationFails_NextUplo 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); @@ -259,9 +268,13 @@ public async Task UploadAsync_AllWaitersCancelBeforeInitializationFails_NextUplo initTcs.SetException(new RequestFailedException(503, "Service unavailable")); - // Give the shared initialization task's own continuation a chance to run and self-heal - // the cache, even though no caller is left waiting on it. - await Task.Delay(100); + // 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. From b4f8306cc3500cd9682b31027b53a0e86adb1bc9 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 27 Jul 2026 11:36:53 -0700 Subject: [PATCH 9/9] Retry upload after container deletion Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 01dc5b7c-c742-45b7-ae8c-5cda9bcfbd2c --- .../PayloadStore/BlobPayloadStore.cs | 84 +++++++++++-------- .../BlobPayloadStoreTests.cs | 17 ++-- 2 files changed, 55 insertions(+), 46 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs index dc8b2abb..76a7a7c7 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -151,51 +151,61 @@ public override async Task UploadAsync(string payLoad, CancellationToken // 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. - // Keep the specific initializer instance this upload used so the ContainerNotFound - // recovery below can invalidate it precisely (see the catch block). - Lazy containerInitializer = await this.EnsureContainerExistsAsync(cancellationToken).ConfigureAwait(false); - - try + // 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) { - if (this.options.CompressionEnabled) + // 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 { - BlobOpenWriteOptions writeOptions = new() + if (this.options.CompressionEnabled) { - 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); + 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); - // await payloadStream.CopyToAsync(compressedBlobStream, bufferSize: DefaultCopyBufferSize, cancellationToken); - await WritePayloadAsync(payloadBuffer, compressedBlobStream, cancellationToken); - await compressedBlobStream.FlushAsync(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); + } } - else + catch (RequestFailedException ex) when ( + retryAfterContainerNotFound && + ex.ErrorCode == BlobErrorCode.ContainerNotFound) { - 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); + // 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; } - } - catch (RequestFailedException ex) when (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 the next upload attempts to - // recreate the container, keeping deliberate deletion recoverable. CompareExchange - // against the specific initializer this upload used ensures a stale failure can - // never clobber a newer initializer already published by another, faster-recovering - // concurrent upload that detected and recovered from the same deletion. - _ = Interlocked.CompareExchange(ref this.containerInitializer, null, containerInitializer); - throw; - } - return EncodeToken(this.containerClient.Name, blobName); + return EncodeToken(this.containerClient.Name, blobName); + } } /// diff --git a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs index 16dc2fdc..5c8576c2 100644 --- a/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs +++ b/test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs @@ -155,23 +155,21 @@ public async Task UploadAsync_ContainerDeletedAfterInitialization_ResetsCacheAnd .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(new MemoryStream()); + .ReturnsAsync(successfulWriteStream); 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); + string token = await store.UploadAsync("payload", CancellationToken.None); - // Assert: the container-not-found failure reset the cache, so the second upload - // recreated the container instead of assuming it still existed. - secondToken.Should().NotBeNullOrEmpty(); + // 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(), @@ -196,10 +194,11 @@ public async Task UploadAsync_UnrelatedStorageError_PropagatesWithoutInvalidatin .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(new MemoryStream()); + .ReturnsAsync(retryWriteStream); containerClientMock.Setup(c => c.GetBlobClient(It.IsAny())).Returns(blobClientMock.Object); BlobPayloadStore store = new(new LargePayloadStorageOptions(), containerClientMock.Object);