Cache Azure Blob container initialization to avoid per-upload CreateIfNotExistsAsync#785
Open
berndverst wants to merge 8 commits into
Open
Cache Azure Blob container initialization to avoid per-upload CreateIfNotExistsAsync#785berndverst wants to merge 8 commits into
berndverst wants to merge 8 commits into
Conversation
…fNotExistsAsync 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
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
| blobClientMock | ||
| .SetupSequence(b => b.OpenWriteAsync(It.IsAny<bool>(), It.IsAny<BlobOpenWriteOptions>(), It.IsAny<CancellationToken>())) | ||
| .ThrowsAsync(new RequestFailedException(404, "The specified container does not exist.", BlobErrorCode.ContainerNotFound.ToString(), null)) | ||
| .ReturnsAsync(new MemoryStream()); |
| blobClientMock | ||
| .SetupSequence(b => b.OpenWriteAsync(It.IsAny<bool>(), It.IsAny<BlobOpenWriteOptions>(), It.IsAny<CancellationToken>())) | ||
| .ThrowsAsync(new RequestFailedException(500, "Internal server error")) | ||
| .ReturnsAsync(new MemoryStream()); |
Contributor
There was a problem hiding this comment.
Pull request overview
This PR improves the Azure Blob payload externalization path by eliminating a per-upload CreateIfNotExistsAsync call, replacing it with a cached “single-flight” container initialization so repeated uploads avoid an extra storage transaction. It also adds a dedicated unit test project to validate initialization caching, cancellation isolation, failure retry, and cache reset behavior.
Changes:
- Cache container initialization in
BlobPayloadStoreto avoid callingCreateIfNotExistsAsyncon every upload. - Reset the cached initialization state when a write fails with
ContainerNotFoundto enable recovery after out-of-band container deletion. - Add a new
AzureBlobPayloads.Testsproject with focused unit tests for concurrency, failure, cancellation, and error propagation scenarios.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs |
Adds cached/single-flight container initialization and cache reset on ContainerNotFound. |
test/Extensions/AzureBlobPayloads.Tests/BlobPayloadStoreTests.cs |
Introduces unit tests for initialization caching behavior and failure/cancellation scenarios. |
test/Extensions/AzureBlobPayloads.Tests/Usings.cs |
Adds global test usings for FluentAssertions/Moq/Xunit. |
test/Extensions/AzureBlobPayloads.Tests/AzureBlobPayloads.Tests.csproj |
Adds new test project referencing Azure Blob payloads extension. |
Microsoft.DurableTask.sln |
Wires the new test project into the solution under an Extensions folder. |
Comment on lines
+289
to
+294
| Task BeginContainerInitialization() | ||
| { | ||
| Task newInitializationTask = this.CreateContainerIfNotExistsAsync(); | ||
| return Interlocked.CompareExchange(ref this.containerInitializationTask, newInitializationTask, null) | ||
| ?? newInitializationTask; | ||
| } |
Comment on lines
+38
to
+42
| // Act: fire many concurrent uploads against a fresh (uninitialized) store. | ||
| Task<string>[] uploads = Enumerable.Range(0, 10) | ||
| .Select(_ => store.UploadAsync("payload", CancellationToken.None)) | ||
| .ToArray(); | ||
| await Task.WhenAll(uploads); |
Address 3 medium-severity concurrency issues found in code review of PR #785: 1. True single-flight: PublishNewInitializer now atomically publishes an unstarted Lazy<Task> gate via CompareExchange before any real work starts, so racing first-time callers can never each independently trigger their own CreateIfNotExistsAsync call. Lazy<Task> 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<Task> 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
Comment on lines
+201
to
+224
| static async Task WaitForInitializationAsync(Task initializationTask, CancellationToken cancellationToken) | ||
| { | ||
| #if NETSTANDARD2_0 | ||
| if (!cancellationToken.CanBeCanceled || initializationTask.IsCompleted) | ||
| { | ||
| await initializationTask.ConfigureAwait(false); | ||
| return; | ||
| } | ||
|
|
||
| TaskCompletionSource<bool> cancellationTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); | ||
| using (cancellationToken.Register(static state => ((TaskCompletionSource<bool>)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 | ||
| } |
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
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (1)
src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs:236
- In the non-NETSTANDARD2_0 branch, awaiting
initializationTask.WaitAsync(cancellationToken)can throwOperationCanceledExceptionwithout ever observing the underlyinginitializationTask. If all callers cancel before the shared initialization later faults, the fault may go unobserved and can surface viaTaskScheduler.UnobservedTaskExceptionon finalization. Consider mirroring the NETSTANDARD2_0 cancellation path by attaching a fault-observing continuation when cancellation happens first.
await initializationTask.WaitAsync(cancellationToken).ConfigureAwait(false);
#endif
…nitializer 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
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<Exception>? 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
Comment on lines
+342
to
+346
| _ = task.ContinueWith( | ||
| t => this.OnInitializationFaultObserved?.Invoke(t.Exception!), | ||
| CancellationToken.None, | ||
| TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, | ||
| TaskScheduler.Default); |
…olling 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<Exception> 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
…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
…al 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
Comment on lines
+186
to
+195
| 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 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; |
Comment on lines
+11
to
+16
| /// <summary> | ||
| /// Unit tests for <see cref="BlobPayloadStore"/>'s container-initialization caching behavior | ||
| /// (see https://github.com/microsoft/durabletask-dotnet/issues/771). | ||
| /// </summary> | ||
| public class BlobPayloadStoreTests | ||
| { |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
BlobPayloadStore.UploadAsyncpreviously calledCreateIfNotExistsAsyncon every upload, adding an extra storage request/transaction to each call even after the container was already known to exist.This change caches the container-initialization result in a single-flight task so
CreateIfNotExistsAsyncis issued at most once perBlobPayloadStoreinstance, while preserving all existing safety guarantees:Task(viaInterlocked.CompareExchange), so only oneCreateIfNotExistsAsynccall is ever made for the first upload burst.CancellationToken(via aWaitAsync/Task.WhenAnybased helper) without canceling the underlying shared operation for other callers still waiting on it.RequestFailedExceptionwhereErrorCode == BlobErrorCode.ContainerNotFound(e.g. the container was deleted out-of-band after initialization), the cache is reset so the next upload recreates the container automatically.No public API changes — this is purely an internal implementation detail of
BlobPayloadStore.Testing
Added
test/Extensions/AzureBlobPayloads.Testswith focused unit tests covering:CreateIfNotExistsAsynccall (single-flight).ContainerNotFound) resets the cache and triggers recreation on the next upload.All 5 new tests pass; the modified project builds cleanly (0 warnings, 0 errors) across all target frameworks (
netstandard2.0,net6.0,net8.0,net10.0).This change is independent of any other in-flight performance work and is scoped entirely to
src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs(plus the new test project).Fixes #771