From 4a8a0aab239226cd4a3d3532eb4c98cec57f32be Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 15:12:24 -0700 Subject: [PATCH 01/10] Bound payload operation concurrency for fan-out messages AzureBlobPayloadsSideCarInterceptor previously awaited each Azure Blob payload externalization/resolution sequentially per field, which adds serial latency proportional to the number of independent payloads in a message (e.g. fan-out orchestrator actions, history chunks, entity batches). This change groups independent operations per message and runs them with a bounded concurrency of 8 simultaneous requests via a new RunWithBoundedConcurrencyAsync helper, instead of an unbounded Task.WhenAll, to avoid tripping Azure Storage account-level throttling. - Preserves protobuf field assignment correctness and collection ordering (each operation closure captures and assigns to its own distinct field/element). - Preserves cancellation: checked before starting any not-yet-started operation. - Preserves first-failure semantics: once an operation fails, no new operations are started, and the first exception propagates via Task.WhenAll, matching prior sequential-await behavior relied upon by callers (e.g. conversion to TaskFailureDetails). - No public API changes; all changes are internal to the sealed interceptor class. Adds focused unit tests covering resolve/externalize ordering, concurrency bound, cancellation, and permanent-failure conversion. Fixes #775 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8171f40b-65d8-4cd5-b14c-59895e8ee8e8 --- .../AzureBlobPayloadsSideCarInterceptor.cs | 222 ++++++-- ...zureBlobPayloadsSideCarInterceptorTests.cs | 489 ++++++++++++++++++ 2 files changed, 658 insertions(+), 53 deletions(-) create mode 100644 test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs index 24236f5b..9a6f0dc9 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -15,6 +15,13 @@ namespace Microsoft.DurableTask; public sealed class AzureBlobPayloadsSideCarInterceptor(PayloadStore payloadStore, LargePayloadStorageOptions options) : PayloadInterceptor(payloadStore, options) { + // Conservative cap on simultaneous Azure Blob uploads/downloads for a single message (e.g. a + // fan-out orchestrator response, a history chunk, or a batch of entity operations). Bounding + // concurrency lets independent payload operations overlap -- avoiding additive per-field + // latency -- without issuing an unbounded burst of requests that could trip Azure Storage + // account-level throttling (see https://aka.ms/azure-storage-scalability-targets). + const int MaxConcurrentPayloadOperations = 8; + /// protected override async Task ExternalizeRequestPayloadsAsync(TRequest request, CancellationToken cancellation) { @@ -112,23 +119,37 @@ protected override async Task ResolveResponsePayloadsAsync(TResponse switch (response) { case P.GetInstanceResponse r when r.OrchestrationState is { } s: - s.Input = await this.MaybeResolveAsync(s.Input, cancellation); - s.Output = await this.MaybeResolveAsync(s.Output, cancellation); - s.CustomStatus = await this.MaybeResolveAsync(s.CustomStatus, cancellation); + await RunWithBoundedConcurrencyAsync( + [ + async () => s.Input = await this.MaybeResolveAsync(s.Input, cancellation), + async () => s.Output = await this.MaybeResolveAsync(s.Output, cancellation), + async () => s.CustomStatus = await this.MaybeResolveAsync(s.CustomStatus, cancellation), + ], + cancellation); break; case P.HistoryChunk c when c.Events != null: - foreach (P.HistoryEvent e in c.Events) { - await this.ResolveEventPayloadsAsync(e, cancellation); + List> operations = []; + foreach (P.HistoryEvent e in c.Events) + { + operations.Add(() => this.ResolveEventPayloadsAsync(e, cancellation)); + } + + await RunWithBoundedConcurrencyAsync(operations, cancellation); } break; case P.QueryInstancesResponse r: - foreach (P.OrchestrationState s in r.OrchestrationState) { - s.Input = await this.MaybeResolveAsync(s.Input, cancellation); - s.Output = await this.MaybeResolveAsync(s.Output, cancellation); - s.CustomStatus = await this.MaybeResolveAsync(s.CustomStatus, cancellation); + List> operations = []; + foreach (P.OrchestrationState s in r.OrchestrationState) + { + operations.Add(async () => s.Input = await this.MaybeResolveAsync(s.Input, cancellation)); + operations.Add(async () => s.Output = await this.MaybeResolveAsync(s.Output, cancellation)); + operations.Add(async () => s.CustomStatus = await this.MaybeResolveAsync(s.CustomStatus, cancellation)); + } + + await RunWithBoundedConcurrencyAsync(operations, cancellation); } break; @@ -136,57 +157,68 @@ protected override async Task ResolveResponsePayloadsAsync(TResponse em.SerializedState = await this.MaybeResolveAsync(em.SerializedState, cancellation); break; case P.QueryEntitiesResponse r: - foreach (P.EntityMetadata em in r.Entities) { - em.SerializedState = await this.MaybeResolveAsync(em.SerializedState, cancellation); + List> operations = []; + foreach (P.EntityMetadata em in r.Entities) + { + operations.Add(async () => em.SerializedState = await this.MaybeResolveAsync(em.SerializedState, cancellation)); + } + + await RunWithBoundedConcurrencyAsync(operations, cancellation); } break; case P.WorkItem wi: - // Resolve activity input - if (wi.ActivityRequest is { } ar) { - ar.Input = await this.MaybeResolveAsync(ar.Input, cancellation); - } + List> operations = []; - // Resolve orchestration input embedded in ExecutionStarted event and external events - if (wi.OrchestratorRequest is { } or) - { - foreach (P.HistoryEvent? e in or.PastEvents) + // Resolve activity input + if (wi.ActivityRequest is { } ar) { - await this.ResolveEventPayloadsAsync(e, cancellation); + operations.Add(async () => ar.Input = await this.MaybeResolveAsync(ar.Input, cancellation)); } - foreach (P.HistoryEvent? e in or.NewEvents) + // Resolve orchestration input embedded in ExecutionStarted event and external events + if (wi.OrchestratorRequest is { } or) { - await this.ResolveEventPayloadsAsync(e, cancellation); + foreach (P.HistoryEvent? e in or.PastEvents) + { + operations.Add(() => this.ResolveEventPayloadsAsync(e, cancellation)); + } + + foreach (P.HistoryEvent? e in or.NewEvents) + { + operations.Add(() => this.ResolveEventPayloadsAsync(e, cancellation)); + } } - } - // Resolve entity V1 batch request (OperationRequest inputs and entity state) - if (wi.EntityRequest is { } er1) - { - er1.EntityState = await this.MaybeResolveAsync(er1.EntityState, cancellation); - if (er1.Operations != null) + // Resolve entity V1 batch request (OperationRequest inputs and entity state) + if (wi.EntityRequest is { } er1) { - foreach (P.OperationRequest op in er1.Operations) + operations.Add(async () => er1.EntityState = await this.MaybeResolveAsync(er1.EntityState, cancellation)); + if (er1.Operations != null) { - op.Input = await this.MaybeResolveAsync(op.Input, cancellation); + foreach (P.OperationRequest op in er1.Operations) + { + operations.Add(async () => op.Input = await this.MaybeResolveAsync(op.Input, cancellation)); + } } } - } - // Resolve entity V2 request (history-based operation requests and entity state) - if (wi.EntityRequestV2 is { } er2) - { - er2.EntityState = await this.MaybeResolveAsync(er2.EntityState, cancellation); - if (er2.OperationRequests != null) + // Resolve entity V2 request (history-based operation requests and entity state) + if (wi.EntityRequestV2 is { } er2) { - foreach (P.HistoryEvent opEvt in er2.OperationRequests) + operations.Add(async () => er2.EntityState = await this.MaybeResolveAsync(er2.EntityState, cancellation)); + if (er2.OperationRequests != null) { - await this.ResolveEventPayloadsAsync(opEvt, cancellation); + foreach (P.HistoryEvent opEvt in er2.OperationRequests) + { + operations.Add(() => this.ResolveEventPayloadsAsync(opEvt, cancellation)); + } } } + + await RunWithBoundedConcurrencyAsync(operations, cancellation); } break; @@ -195,60 +227,64 @@ protected override async Task ResolveResponsePayloadsAsync(TResponse async Task ExternalizeOrchestratorResponseAsync(P.OrchestratorResponse r, CancellationToken cancellation) { - r.CustomStatus = await this.MaybeExternalizeAsync(r.CustomStatus, cancellation); + List> operations = [async () => r.CustomStatus = await this.MaybeExternalizeAsync(r.CustomStatus, cancellation)]; + foreach (P.OrchestratorAction a in r.Actions) { if (a.CompleteOrchestration is { } complete) { - complete.Result = await this.MaybeExternalizeAsync(complete.Result, cancellation); - complete.Details = await this.MaybeExternalizeAsync(complete.Details, cancellation); + operations.Add(async () => complete.Result = await this.MaybeExternalizeAsync(complete.Result, cancellation)); + operations.Add(async () => complete.Details = await this.MaybeExternalizeAsync(complete.Details, cancellation)); } if (a.TerminateOrchestration is { } term) { - term.Reason = await this.MaybeExternalizeAsync(term.Reason, cancellation); + operations.Add(async () => term.Reason = await this.MaybeExternalizeAsync(term.Reason, cancellation)); } if (a.ScheduleTask is { } schedule) { - schedule.Input = await this.MaybeExternalizeAsync(schedule.Input, cancellation); + operations.Add(async () => schedule.Input = await this.MaybeExternalizeAsync(schedule.Input, cancellation)); } if (a.CreateSubOrchestration is { } sub) { - sub.Input = await this.MaybeExternalizeAsync(sub.Input, cancellation); + operations.Add(async () => sub.Input = await this.MaybeExternalizeAsync(sub.Input, cancellation)); } if (a.SendEvent is { } sendEvt) { - sendEvt.Data = await this.MaybeExternalizeAsync(sendEvt.Data, cancellation); + operations.Add(async () => sendEvt.Data = await this.MaybeExternalizeAsync(sendEvt.Data, cancellation)); } if (a.SendEntityMessage is { } entityMsg) { if (entityMsg.EntityOperationSignaled is { } sig) { - sig.Input = await this.MaybeExternalizeAsync(sig.Input, cancellation); + operations.Add(async () => sig.Input = await this.MaybeExternalizeAsync(sig.Input, cancellation)); } if (entityMsg.EntityOperationCalled is { } called) { - called.Input = await this.MaybeExternalizeAsync(called.Input, cancellation); + operations.Add(async () => called.Input = await this.MaybeExternalizeAsync(called.Input, cancellation)); } } } + + await RunWithBoundedConcurrencyAsync(operations, cancellation); } async Task ExternalizeEntityBatchResultAsync(P.EntityBatchResult r, CancellationToken cancellation) { - r.EntityState = await this.MaybeExternalizeAsync(r.EntityState, cancellation); + List> operations = [async () => r.EntityState = await this.MaybeExternalizeAsync(r.EntityState, cancellation)]; + if (r.Results != null) { foreach (P.OperationResult result in r.Results) { if (result.Success is { } success) { - success.Result = await this.MaybeExternalizeAsync(success.Result, cancellation); + operations.Add(async () => success.Result = await this.MaybeExternalizeAsync(success.Result, cancellation)); } } } @@ -259,27 +295,32 @@ async Task ExternalizeEntityBatchResultAsync(P.EntityBatchResult r, Cancellation { if (action.SendSignal is { } sendSig) { - sendSig.Input = await this.MaybeExternalizeAsync(sendSig.Input, cancellation); + operations.Add(async () => sendSig.Input = await this.MaybeExternalizeAsync(sendSig.Input, cancellation)); } if (action.StartNewOrchestration is { } start) { - start.Input = await this.MaybeExternalizeAsync(start.Input, cancellation); + operations.Add(async () => start.Input = await this.MaybeExternalizeAsync(start.Input, cancellation)); } } } + + await RunWithBoundedConcurrencyAsync(operations, cancellation); } async Task ExternalizeEntityBatchRequestAsync(P.EntityBatchRequest r, CancellationToken cancellation) { - r.EntityState = await this.MaybeExternalizeAsync(r.EntityState, cancellation); + List> operations = [async () => r.EntityState = await this.MaybeExternalizeAsync(r.EntityState, cancellation)]; + if (r.Operations != null) { foreach (P.OperationRequest op in r.Operations) { - op.Input = await this.MaybeExternalizeAsync(op.Input, cancellation); + operations.Add(async () => op.Input = await this.MaybeExternalizeAsync(op.Input, cancellation)); } } + + await RunWithBoundedConcurrencyAsync(operations, cancellation); } async Task ResolveEventPayloadsAsync(P.HistoryEvent e, CancellationToken cancellation) @@ -410,6 +451,81 @@ async Task ResolveEventPayloadsAsync(P.HistoryEvent e, CancellationToken cancell } } + /// + /// Runs a set of independent Azure Blob payload externalization/resolution operations with + /// bounded concurrency, instead of either awaiting them one at a time (additive latency) or + /// firing them all at once via an unbounded + /// (risking Azure Storage throttling for messages with many payloads). Each delegate is + /// expected to assign its result to a distinct protobuf field/element, so the relative + /// completion order between operations does not affect correctness -- only the resulting + /// number of simultaneously in-flight requests is bounded. + /// + /// + /// Cancellation is honored before starting any operation not already in flight. If any + /// operation throws (e.g. for an oversized payload, or + /// a non-retriable ), that exception propagates the same + /// way it would have from a sequential await chain, preserving the existing first-failure + /// handling performed by callers (e.g. converting the failure into a + /// completion). + /// + /// The independent operations to run. + /// Cancellation token. + static async Task RunWithBoundedConcurrencyAsync(IReadOnlyList> operations, CancellationToken cancellation) + { + if (operations.Count == 0) + { + return; + } + + if (operations.Count == 1) + { + // Fast path: avoid semaphore/list overhead for the overwhelmingly common single-field case. + await operations[0](); + return; + } + + using SemaphoreSlim throttle = new(MaxConcurrentPayloadOperations, MaxConcurrentPayloadOperations); + List inFlight = new(operations.Count); + Exception? firstFailure = null; + + foreach (Func operation in operations) + { + cancellation.ThrowIfCancellationRequested(); + + // Once an earlier operation has failed permanently, stop issuing new Azure Storage + // requests. Operations already started are left to drain below rather than + // cancelled, since they may already have side effects (e.g. an in-flight upload). + if (Volatile.Read(ref firstFailure) != null) + { + break; + } + + await throttle.WaitAsync(cancellation); + inFlight.Add(TrackAsync(operation)); + } + + // Await propagates the first exception encountered (matching the prior sequential + // await-per-field behavior), after allowing every started operation to complete. + await Task.WhenAll(inFlight); + + async Task TrackAsync(Func operation) + { + try + { + await operation(); + } + catch (Exception ex) + { + Interlocked.CompareExchange(ref firstFailure, ex, null); + throw; + } + finally + { + throttle.Release(); + } + } + } + /// /// Determines whether an exception represents a permanent storage failure that will never /// succeed on retry, such as payload exceeding the configured maximum or 4xx HTTP errors diff --git a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs new file mode 100644 index 00000000..570dfe06 --- /dev/null +++ b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs @@ -0,0 +1,489 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Reflection; +using FluentAssertions; +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Grpc.Tests; + +/// +/// Focused unit tests for 's bounded-concurrency +/// payload externalization/resolution logic (see GitHub issue #775: "Performance: avoid serial +/// Azure Blob payload operations for fan-out messages"). These tests invoke the interceptor's +/// protected ExternalizeRequestPayloadsAsync/ResolveResponsePayloadsAsync methods directly via +/// reflection against an in-memory test double, without requiring a +/// running gRPC sidecar. +/// +public sealed class AzureBlobPayloadsSideCarInterceptorTests +{ + static readonly MethodInfo ExternalizeMethodDefinition = typeof(AzureBlobPayloadsSideCarInterceptor) + .GetMethod("ExternalizeRequestPayloadsAsync", BindingFlags.Instance | BindingFlags.NonPublic)!; + + static readonly MethodInfo ResolveMethodDefinition = typeof(AzureBlobPayloadsSideCarInterceptor) + .GetMethod("ResolveResponsePayloadsAsync", BindingFlags.Instance | BindingFlags.NonPublic)!; + + [Fact] + public async Task ResolveResponsePayloadsAsync_HistoryChunk_ResolvesEventsInOrderWithBoundedConcurrency() + { + // Arrange: 20 independent events, enough to observe 8-way bounded concurrency overlap. + const int eventCount = 20; + TrackingPayloadStore store = new(delay: TimeSpan.FromMilliseconds(50)); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + + P.HistoryChunk chunk = new(); + string[] expected = new string[eventCount]; + for (int i = 0; i < eventCount; i++) + { + expected[i] = $"payload-{i}"; + chunk.Events.Add(new P.HistoryEvent + { + EventId = i, + ExecutionStarted = new P.ExecutionStartedEvent { Input = store.Seed(expected[i]) }, + }); + } + + // Act + await ResolveAsync(interceptor, chunk, CancellationToken.None); + + // Assert: every event resolved to the correct value, in the correct position. + for (int i = 0; i < eventCount; i++) + { + chunk.Events[i].ExecutionStarted.Input.Should().Be(expected[i]); + } + + store.DownloadCount.Should().Be(eventCount); + store.MaxObservedConcurrency.Should().BeGreaterThan(1, "independent payload operations should overlap, not run strictly sequentially"); + store.MaxObservedConcurrency.Should().BeLessOrEqualTo(8, "concurrency must be bounded to avoid Azure Storage throttling"); + } + + [Fact] + public async Task ResolveResponsePayloadsAsync_QueryInstancesResponse_ResolvesEachInstanceIndependently() + { + // Arrange + const int instanceCount = 12; + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + + P.QueryInstancesResponse response = new(); + string[] expectedInputs = new string[instanceCount]; + string[] expectedOutputs = new string[instanceCount]; + for (int i = 0; i < instanceCount; i++) + { + expectedInputs[i] = $"input-{i}"; + expectedOutputs[i] = $"output-{i}"; + response.OrchestrationState.Add(new P.OrchestrationState + { + InstanceId = $"instance-{i}", + Input = store.Seed(expectedInputs[i]), + Output = store.Seed(expectedOutputs[i]), + }); + } + + // Act + await ResolveAsync(interceptor, response, CancellationToken.None); + + // Assert: no cross-instance mixups under concurrent resolution. + for (int i = 0; i < instanceCount; i++) + { + response.OrchestrationState[i].Input.Should().Be(expectedInputs[i]); + response.OrchestrationState[i].Output.Should().Be(expectedOutputs[i]); + } + } + + [Fact] + public async Task ResolveResponsePayloadsAsync_WorkItemOrchestratorRequest_ResolvesPastAndNewEventsInOrder() + { + // Arrange + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + + P.WorkItem workItem = new() + { + OrchestratorRequest = new P.OrchestratorRequest { InstanceId = "instance-1" }, + }; + + string[] pastExpected = ["past-0", "past-1", "past-2"]; + string[] newExpected = ["new-0", "new-1"]; + + foreach (string value in pastExpected) + { + workItem.OrchestratorRequest.PastEvents.Add(new P.HistoryEvent + { + TaskScheduled = new P.TaskScheduledEvent { Input = store.Seed(value) }, + }); + } + + foreach (string value in newExpected) + { + workItem.OrchestratorRequest.NewEvents.Add(new P.HistoryEvent + { + TaskCompleted = new P.TaskCompletedEvent { Result = store.Seed(value) }, + }); + } + + // Act + await ResolveAsync(interceptor, workItem, CancellationToken.None); + + // Assert: past and new events resolve independently without swapping values across lists/positions. + for (int i = 0; i < pastExpected.Length; i++) + { + workItem.OrchestratorRequest.PastEvents[i].TaskScheduled.Input.Should().Be(pastExpected[i]); + } + + for (int i = 0; i < newExpected.Length; i++) + { + workItem.OrchestratorRequest.NewEvents[i].TaskCompleted.Result.Should().Be(newExpected[i]); + } + } + + [Fact] + public async Task ResolveResponsePayloadsAsync_WorkItemEntityRequestV1_ResolvesEntityStateAndOperationsInOrder() + { + // Arrange + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + + P.WorkItem workItem = new() + { + EntityRequest = new P.EntityBatchRequest { InstanceId = "entity-1", EntityState = store.Seed("entity-state") }, + }; + + const int operationCount = 6; + string[] expected = new string[operationCount]; + for (int i = 0; i < operationCount; i++) + { + expected[i] = $"op-{i}"; + workItem.EntityRequest.Operations.Add(new P.OperationRequest + { + Operation = "op", + RequestId = $"req-{i}", + Input = store.Seed(expected[i]), + }); + } + + // Act + await ResolveAsync(interceptor, workItem, CancellationToken.None); + + // Assert + workItem.EntityRequest.EntityState.Should().Be("entity-state"); + for (int i = 0; i < operationCount; i++) + { + workItem.EntityRequest.Operations[i].Input.Should().Be(expected[i]); + } + } + + [Fact] + public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_ExternalizesActionsInOrderWithBoundedConcurrency() + { + // Arrange: 16 independent actions, enough to observe 8-way bounded concurrency overlap. + const int actionCount = 16; + TrackingPayloadStore store = new(delay: TimeSpan.FromMilliseconds(50)); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + + P.OrchestratorResponse response = new() { InstanceId = "instance-1", CustomStatus = "status-0" }; + string[] expectedInputs = new string[actionCount]; + for (int i = 0; i < actionCount; i++) + { + expectedInputs[i] = $"action-input-{i}"; + response.Actions.Add(new P.OrchestratorAction + { + Id = i, + ScheduleTask = new P.ScheduleTaskAction { Name = $"activity-{i}", Input = expectedInputs[i] }, + }); + } + + // Act + await ExternalizeAsync(interceptor, response, CancellationToken.None); + + // Assert: every action's input externalized to a distinct token round-tripping to the correct value. + store.GetUploadedValue(response.CustomStatus).Should().Be("status-0"); + for (int i = 0; i < actionCount; i++) + { + string token = response.Actions[i].ScheduleTask.Input; + token.Should().NotBe(expectedInputs[i]); + store.GetUploadedValue(token).Should().Be(expectedInputs[i]); + } + + store.MaxObservedConcurrency.Should().BeGreaterThan(1, "independent payload operations should overlap, not run strictly sequentially"); + store.MaxObservedConcurrency.Should().BeLessOrEqualTo(8, "concurrency must be bounded to avoid Azure Storage throttling"); + } + + [Fact] + public async Task ExternalizeRequestPayloadsAsync_EntityBatchRequest_ExternalizesOperationsInOrder() + { + // Arrange + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + + const int operationCount = 10; + P.EntityBatchRequest request = new() { InstanceId = "entity-1", EntityState = "state-0" }; + string[] expectedInputs = new string[operationCount]; + for (int i = 0; i < operationCount; i++) + { + expectedInputs[i] = $"op-input-{i}"; + request.Operations.Add(new P.OperationRequest { Operation = "op", RequestId = $"req-{i}", Input = expectedInputs[i] }); + } + + // Act + await ExternalizeAsync(interceptor, request, CancellationToken.None); + + // Assert + store.GetUploadedValue(request.EntityState).Should().Be("state-0"); + for (int i = 0; i < operationCount; i++) + { + string token = request.Operations[i].Input; + token.Should().NotBe(expectedInputs[i]); + store.GetUploadedValue(token).Should().Be(expectedInputs[i]); + } + } + + [Fact] + public async Task ExternalizeRequestPayloadsAsync_EntityBatchResult_ExternalizesResultsAndActionsInOrder() + { + // Arrange + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + + P.EntityBatchResult result = new() { EntityState = "state-final" }; + + const int resultCount = 6; + string[] expectedResults = new string[resultCount]; + for (int i = 0; i < resultCount; i++) + { + expectedResults[i] = $"success-result-{i}"; + result.Results.Add(new P.OperationResult { Success = new P.OperationResultSuccess { Result = expectedResults[i] } }); + } + + result.Actions.Add(new P.OperationAction + { + Id = 1, + SendSignal = new P.SendSignalAction { InstanceId = "target-1", Name = "signal", Input = "signal-input" }, + }); + result.Actions.Add(new P.OperationAction + { + Id = 2, + StartNewOrchestration = new P.StartNewOrchestrationAction { InstanceId = "target-2", Name = "orch", Input = "start-input" }, + }); + + // Act + await ExternalizeAsync(interceptor, result, CancellationToken.None); + + // Assert + store.GetUploadedValue(result.EntityState).Should().Be("state-final"); + for (int i = 0; i < resultCount; i++) + { + store.GetUploadedValue(result.Results[i].Success.Result).Should().Be(expectedResults[i]); + } + + store.GetUploadedValue(result.Actions[0].SendSignal.Input).Should().Be("signal-input"); + store.GetUploadedValue(result.Actions[1].StartNewOrchestration.Input).Should().Be("start-input"); + } + + [Fact] + public async Task ExternalizeRequestPayloadsAsync_ActivityResponse_PermanentFailure_ConvertsToFailureDetails() + { + // Arrange: MaxPayloadBytes = 10 so the oversized result triggers a permanent PayloadStorageException. + TrackingPayloadStore store = new(); + LargePayloadStorageOptions options = new() { ThresholdBytes = 1, MaxPayloadBytes = 10 }; + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, options); + + P.ActivityResponse response = new() { InstanceId = "instance-1", TaskId = 1, Result = new string('y', 100) }; + + // Act + await ExternalizeAsync(interceptor, response, CancellationToken.None); + + // Assert: activity result replaced with non-retriable failure details instead of throwing/abandoning the work item. + response.Result.Should().BeNullOrEmpty(); + response.FailureDetails.Should().NotBeNull(); + response.FailureDetails.ErrorType.Should().Be(typeof(PayloadStorageException).FullName); + response.FailureDetails.IsNonRetriable.Should().BeTrue(); + store.UploadCount.Should().Be(0, "the oversized payload should fail before ever reaching the payload store"); + } + + [Fact] + public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_PermanentFailureInOneAction_ReplacesActionsWithFailedCompletion() + { + // Arrange: MaxPayloadBytes = 10 so the second action's oversized input triggers a permanent failure. + TrackingPayloadStore store = new(); + LargePayloadStorageOptions options = new() { ThresholdBytes = 1, MaxPayloadBytes = 10 }; + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, options); + + P.OrchestratorResponse response = new() { InstanceId = "instance-1", CustomStatus = "status" }; + response.Actions.Add(new P.OrchestratorAction + { + Id = 1, + ScheduleTask = new P.ScheduleTaskAction { Name = "activity-1", Input = "short" }, + }); + response.Actions.Add(new P.OrchestratorAction + { + Id = 2, + ScheduleTask = new P.ScheduleTaskAction { Name = "activity-2", Input = new string('x', 100) }, + }); + + // Act + await ExternalizeAsync(interceptor, response, CancellationToken.None); + + // Assert: entire response replaced with a single Failed completion action, per the pre-existing + // (unrefactored) fallback contract for permanent externalization failures. + response.Actions.Should().HaveCount(1); + P.OrchestratorAction resultAction = response.Actions[0]; + resultAction.CompleteOrchestration.Should().NotBeNull(); + resultAction.CompleteOrchestration.OrchestrationStatus.Should().Be(P.OrchestrationStatus.Failed); + resultAction.CompleteOrchestration.FailureDetails.ErrorType.Should().Be(typeof(PayloadStorageException).FullName); + resultAction.CompleteOrchestration.FailureDetails.IsNonRetriable.Should().BeTrue(); + response.CustomStatus.Should().BeNullOrEmpty(); +#pragma warning disable CS0612 // isPartial/chunkIndex are deprecated but still required for chunked response wire compatibility. + response.IsPartial.Should().BeFalse(); + response.ChunkIndex.Should().BeNull(); +#pragma warning restore CS0612 + } + + [Fact] + public async Task ResolveResponsePayloadsAsync_PreCancelledToken_ThrowsAndPerformsNoStoreCalls() + { + // Arrange + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + + P.HistoryChunk chunk = new(); + for (int i = 0; i < 5; i++) + { + chunk.Events.Add(new P.HistoryEvent + { + EventId = i, + ExecutionStarted = new P.ExecutionStartedEvent { Input = store.Seed($"value-{i}") }, + }); + } + + using CancellationTokenSource cts = new(); + cts.Cancel(); + + // Act + Func act = () => ResolveAsync(interceptor, chunk, cts.Token); + + // Assert: cancellation observed before any operation starts (no store calls made). + await act.Should().ThrowAsync(); + store.DownloadCount.Should().Be(0); + } + + static LargePayloadStorageOptions CreateOptions() => new() { ThresholdBytes = 1 }; + + static Task ExternalizeAsync(AzureBlobPayloadsSideCarInterceptor interceptor, TRequest request, CancellationToken cancellation) + => (Task)ExternalizeMethodDefinition.MakeGenericMethod(typeof(TRequest)).Invoke(interceptor, [request, cancellation])!; + + static Task ResolveAsync(AzureBlobPayloadsSideCarInterceptor interceptor, TResponse response, CancellationToken cancellation) + => (Task)ResolveMethodDefinition.MakeGenericMethod(typeof(TResponse)).Invoke(interceptor, [response, cancellation])!; + + /// + /// In-memory test double that tracks upload/download counts and the + /// high-water mark of concurrently in-flight calls, and optionally introduces an artificial + /// delay to force overlapping calls so bounded concurrency can be observed deterministically. + /// + sealed class TrackingPayloadStore(TimeSpan delay = default) : PayloadStore + { + const string TokenPrefix = "test-blob://"; + + readonly object gate = new(); + readonly Dictionary tokenToValue = []; + int concurrentCalls; + int uploadCount; + int downloadCount; + + public int MaxObservedConcurrency { get; private set; } + + public int UploadCount => this.uploadCount; + + public int DownloadCount => this.downloadCount; + + /// Pre-populates the store with a value and returns its token, for resolve-direction tests. + public string Seed(string value) + { + string token = TokenPrefix + Guid.NewGuid().ToString("N"); + lock (this.gate) + { + this.tokenToValue[token] = value; + } + + return token; + } + + /// Looks up the original value uploaded/seeded for a given token, for externalize-direction tests. + public string GetUploadedValue(string token) + { + lock (this.gate) + { + return this.tokenToValue[token]; + } + } + + public override async Task UploadAsync(string payLoad, CancellationToken cancellationToken) + { + this.EnterCall(); + try + { + if (delay > TimeSpan.Zero) + { + await Task.Delay(delay, cancellationToken); + } + + Interlocked.Increment(ref this.uploadCount); + string token = TokenPrefix + Guid.NewGuid().ToString("N"); + lock (this.gate) + { + this.tokenToValue[token] = payLoad; + } + + return token; + } + finally + { + this.ExitCall(); + } + } + + public override async Task DownloadAsync(string token, CancellationToken cancellationToken) + { + this.EnterCall(); + try + { + if (delay > TimeSpan.Zero) + { + await Task.Delay(delay, cancellationToken); + } + + Interlocked.Increment(ref this.downloadCount); + lock (this.gate) + { + return this.tokenToValue[token]; + } + } + finally + { + this.ExitCall(); + } + } + + public override bool IsKnownPayloadToken(string value) => value.StartsWith(TokenPrefix, StringComparison.Ordinal); + + void EnterCall() + { + lock (this.gate) + { + this.concurrentCalls++; + if (this.concurrentCalls > this.MaxObservedConcurrency) + { + this.MaxObservedConcurrency = this.concurrentCalls; + } + } + } + + void ExitCall() + { + lock (this.gate) + { + this.concurrentCalls--; + } + } + } +} From f5c0be6d0d97757698b8f67e1a2f4a3c7889eebf Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 15:54:33 -0700 Subject: [PATCH 02/10] Fix cancellation/disposal race in bounded-concurrency payload dispatch Two related bugs in RunWithBoundedConcurrencyAsync found during independent review of PR #786: 1. Semaphore disposal race: the bounding SemaphoreSlim was wrapped in using, so a ThrowIfCancellationRequested or WaitAsync(cancellation) throw during the dispatch loop would unwind and dispose it while already-dispatched TrackAsync tasks were still running and would later call Release() in their finally block. This could produce an unobserved ObjectDisposedException, mask a genuine (non-cancellation) PayloadStorageException from an in-flight operation, and risked a protobuf field mutation after the caller had already received a response/exception. Fix: the semaphore is no longer disposed via using. Every dispatched operation is now always drained via try { await Task.WhenAll(inFlight); } finally { throttle.Dispose(); } before the method returns or throws, and TrackAsync no longer rethrows -- it only records the first exception via Interlocked.CompareExchange, so Task.WhenAll can never fault and disposal is only ever reached once every Release() has already run. The captured first failure is re-thrown afterwards via ExceptionDispatchInfo to preserve its original stack trace and first-failure semantics. 2. Post-WaitAsync cancellation gap: the dispatch loop only checked cancellation/failure before calling throttle.WaitAsync(), not after. A slot freed by an in-flight operation could let a new (9th, 10th, ...) operation dispatch even though cancellation/failure had already been observed while waiting for that slot. Fix: added a second check immediately after WaitAsync() succeeds; if cancellation or a failure is now observed, the just-acquired slot is released back and the loop stops without starting that operation. WaitAsync is now called with CancellationToken.None (not the caller's token), since every dispatched operation already observes the same token itself -- avoiding a WaitAsync-triggered unwind that would abandon an already-tracked operation. Added a regression test (RunWithBoundedConcurrencyAsync_CancelledWhileOperationsInFlight_DrainsBeforeReturningAndDoesNotMaskFailure) that dispatches 9 operations (one over the concurrency limit of 8), cancels after the first 8 are genuinely in-flight, and verifies the call surfaces the genuine InvalidOperationException from an in-flight operation (not OperationCanceledException or ObjectDisposedException), that the 9th operation is never dispatched, and that draining completes within a bounded timeout (guards against a deadlock regression). Also addressed CA2016 (explicit CancellationToken.None) and CA1859 (List> instead of IReadOnlyList> parameter, since all call sites already pass a concrete List) surfaced while touching this method. All 11 tests in AzureBlobPayloadsSideCarInterceptorTests pass, and the full Grpc.IntegrationTests suite (153 tests) passes. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8171f40b-65d8-4cd5-b14c-59895e8ee8e8 --- .../AzureBlobPayloadsSideCarInterceptor.cs | 76 +++++++++++++--- ...zureBlobPayloadsSideCarInterceptorTests.cs | 90 +++++++++++++++++++ 2 files changed, 154 insertions(+), 12 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs index 9a6f0dc9..9307ff19 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -1,6 +1,7 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. +using System.Runtime.ExceptionServices; using Azure; using Grpc.Core.Interceptors; @@ -467,10 +468,17 @@ async Task ResolveEventPayloadsAsync(P.HistoryEvent e, CancellationToken cancell /// way it would have from a sequential await chain, preserving the existing first-failure /// handling performed by callers (e.g. converting the failure into a /// completion). + /// + /// Every started operation is always drained (fully awaited) before this method returns or + /// throws -- including when stopping early due to cancellation or a prior failure -- so that + /// (a) the bounding is only disposed once nothing can call + /// on it, and (b) no operation can mutate its target + /// protobuf field after control has already passed back to the caller. + /// /// /// The independent operations to run. /// Cancellation token. - static async Task RunWithBoundedConcurrencyAsync(IReadOnlyList> operations, CancellationToken cancellation) + static async Task RunWithBoundedConcurrencyAsync(List> operations, CancellationToken cancellation) { if (operations.Count == 0) { @@ -484,29 +492,69 @@ static async Task RunWithBoundedConcurrencyAsync(IReadOnlyList> opera return; } - using SemaphoreSlim throttle = new(MaxConcurrentPayloadOperations, MaxConcurrentPayloadOperations); + SemaphoreSlim throttle = new(MaxConcurrentPayloadOperations, MaxConcurrentPayloadOperations); List inFlight = new(operations.Count); Exception? firstFailure = null; foreach (Func operation in operations) { - cancellation.ThrowIfCancellationRequested(); + // Stop dispatching new Azure Storage requests once an earlier operation has failed + // permanently, or cancellation has been requested. Operations already started are + // *not* abandoned here -- they are drained below -- since they may already have side + // effects (e.g. an in-flight upload) and must not touch the semaphore, or mutate + // their target field, after this method has returned control to the caller. + if (Volatile.Read(ref firstFailure) != null || cancellation.IsCancellationRequested) + { + break; + } - // Once an earlier operation has failed permanently, stop issuing new Azure Storage - // requests. Operations already started are left to drain below rather than - // cancelled, since they may already have side effects (e.g. an in-flight upload). - if (Volatile.Read(ref firstFailure) != null) + // Deliberately passing CancellationToken.None (not `cancellation`) to WaitAsync: every + // dispatched operation is itself cancellation-aware (it receives the same token), so a + // slot reliably frees up as soon as an in-flight operation observes cancellation -- + // without risking a WaitAsync-triggered unwind that would abandon an already-tracked + // operation. + await throttle.WaitAsync(CancellationToken.None); + + // Re-check immediately after acquiring the slot: cancellation or a failure may have + // occurred while this operation was queued waiting for a slot to free up. If so, + // release the slot back (this operation never started, so there is nothing to drain + // for it) and stop dispatching -- otherwise a newly-freed slot could let a new + // operation start after cancellation/failure was already observed. + if (Volatile.Read(ref firstFailure) != null || cancellation.IsCancellationRequested) { + throttle.Release(); break; } - await throttle.WaitAsync(cancellation); inFlight.Add(TrackAsync(operation)); } - // Await propagates the first exception encountered (matching the prior sequential - // await-per-field behavior), after allowing every started operation to complete. - await Task.WhenAll(inFlight); + try + { + // Wait for every started operation to finish -- regardless of outcome -- before this + // method returns or throws. TrackAsync below never lets an exception fault this + // Task.WhenAll; it only records it in `firstFailure`, so draining always completes. + await Task.WhenAll(inFlight); + } + finally + { + // Safe only because every TrackAsync call above (including its own `finally`, which + // releases a semaphore slot) has already run to completion by the time + // Task.WhenAll(inFlight) returns. + throttle.Dispose(); + } + + if (firstFailure != null) + { + // Rethrow the first operation failure with its original stack trace preserved, + // matching the prior sequential await-per-field behavior (and never masked by e.g. + // a later cancellation or a disposed-resource exception). + ExceptionDispatchInfo.Capture(firstFailure).Throw(); + } + + // No operation failed, but cancellation may still have cut dispatch short before every + // operation was started; surface that the same way a sequential await chain would have. + cancellation.ThrowIfCancellationRequested(); async Task TrackAsync(Func operation) { @@ -516,8 +564,12 @@ async Task TrackAsync(Func operation) } catch (Exception ex) { + // Preserve first-failure semantics (only the first exception is surfaced, + // matching the prior sequential behavior). The exception is captured -- not + // rethrown -- so Task.WhenAll above never faults, guaranteeing every operation, + // including this finally block, runs to completion before the semaphore is + // disposed. Interlocked.CompareExchange(ref firstFailure, ex, null); - throw; } finally { diff --git a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs index 570dfe06..69d13516 100644 --- a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs +++ b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs @@ -23,6 +23,9 @@ public sealed class AzureBlobPayloadsSideCarInterceptorTests static readonly MethodInfo ResolveMethodDefinition = typeof(AzureBlobPayloadsSideCarInterceptor) .GetMethod("ResolveResponsePayloadsAsync", BindingFlags.Instance | BindingFlags.NonPublic)!; + static readonly MethodInfo RunWithBoundedConcurrencyMethodDefinition = typeof(AzureBlobPayloadsSideCarInterceptor) + .GetMethod("RunWithBoundedConcurrencyAsync", BindingFlags.Static | BindingFlags.NonPublic)!; + [Fact] public async Task ResolveResponsePayloadsAsync_HistoryChunk_ResolvesEventsInOrderWithBoundedConcurrency() { @@ -367,6 +370,90 @@ public async Task ResolveResponsePayloadsAsync_PreCancelledToken_ThrowsAndPerfor store.DownloadCount.Should().Be(0); } + [Fact] + public async Task RunWithBoundedConcurrencyAsync_CancelledWhileOperationsInFlight_DrainsBeforeReturningAndDoesNotMaskFailure() + { + // Arrange: 9 operations -- one more than MaxConcurrentPayloadOperations (8) -- so the + // dispatch loop fills all 8 concurrency slots with genuinely in-flight operations before + // the 9th operation's semaphore wait forces the loop to observe real contention. This is + // the exact race window in which the original bug -- disposing the bounding SemaphoreSlim + // on cancellation before every in-flight operation had released it -- could surface an + // unobserved ObjectDisposedException, mask a genuine (non-cancellation) failure from an + // already-in-flight operation, and let that operation's continuation run after the caller + // had already received a response. + // + // Operation 0 simulates an upload/download that has already committed and is in the + // process of failing for a real, unrelated reason (e.g. a permanent PayloadStorageException) + // -- it does not observe the cancellation token at all, so it only completes once the test + // explicitly releases it, well after cancellation has been requested. Operations 1-6 + // likewise ignore cancellation and simply succeed once released. Operation 7 (the 9th, index + // 8) must never be dispatched at all, since cancellation is requested before a 9th slot ever + // frees up. + TaskCompletionSource releaseGate = new(TaskCreationOptions.RunContinuationsAsynchronously); + int dispatchedBeforeCancel = 0; + bool ninthOperationDispatched = false; + + List> operations = []; + operations.Add(async () => + { + Interlocked.Increment(ref dispatchedBeforeCancel); + await releaseGate.Task; + throw new InvalidOperationException("Simulated permanent storage failure, unrelated to cancellation."); + }); + for (int i = 1; i < 8; i++) + { + operations.Add(async () => + { + Interlocked.Increment(ref dispatchedBeforeCancel); + await releaseGate.Task; + }); + } + + operations.Add(() => + { + // Should never run: cancellation must stop the dispatch loop before this 9th + // operation's semaphore wait can ever be satisfied. + ninthOperationDispatched = true; + return Task.CompletedTask; + }); + + using CancellationTokenSource cts = new(); + + // Act: start the call without awaiting it yet. The first 8 operations dispatch and block + // on `releaseGate` synchronously (no artificial delay is needed since none of them observe + // cancellation), so by the time this line completes, the 9th operation's semaphore wait is + // already the sole blocking point. + Task runTask = RunWithBoundedConcurrencyAsync(operations, cts.Token); + dispatchedBeforeCancel.Should().Be(8, "the first 8 operations should dispatch synchronously before the 9th blocks on the semaphore"); + + cts.Cancel(); + + // Small buffer so that, under the old (buggy) implementation, the cancellation-triggered + // unwind and premature semaphore disposal have already fully happened before the + // already-in-flight operations are released below -- making the demonstration + // unambiguous rather than a race between this test and that unwind. + await Task.Delay(50); + + // Now let the 8 already-in-flight operations complete: operation 0 fails for a genuine, + // cancellation-unrelated reason; operations 1-6 succeed. + releaseGate.SetResult(true); + + // Assert: the overall call surfaces the genuine InvalidOperationException -- not + // OperationCanceledException (which the old implementation would incorrectly surface + // immediately upon cancellation, before operation 0 ever got a chance to fail) and not + // ObjectDisposedException (which the old implementation could surface from an orphaned, + // unobserved background task once operation 0's `finally` tried to release an + // already-disposed semaphore). A bounded wait guards against a deadlock regression. + Func act = () => runTask.WaitAsync(TimeSpan.FromSeconds(10)); + (await act.Should().ThrowExactlyAsync( + "a genuine failure from an already-in-flight operation must never be masked by a concurrent cancellation or disposal race")) + .WithMessage("Simulated permanent storage failure*"); + + // Assert: the 9th operation was never dispatched -- cancellation correctly stopped the + // dispatch loop from starting new operations, without abandoning the 8 already in flight. + ninthOperationDispatched.Should().BeFalse(); + } + static LargePayloadStorageOptions CreateOptions() => new() { ThresholdBytes = 1 }; static Task ExternalizeAsync(AzureBlobPayloadsSideCarInterceptor interceptor, TRequest request, CancellationToken cancellation) @@ -375,6 +462,9 @@ static Task ExternalizeAsync(AzureBlobPayloadsSideCarInterceptor inter static Task ResolveAsync(AzureBlobPayloadsSideCarInterceptor interceptor, TResponse response, CancellationToken cancellation) => (Task)ResolveMethodDefinition.MakeGenericMethod(typeof(TResponse)).Invoke(interceptor, [response, cancellation])!; + static Task RunWithBoundedConcurrencyAsync(IReadOnlyList> operations, CancellationToken cancellation) + => (Task)RunWithBoundedConcurrencyMethodDefinition.Invoke(null, [operations, cancellation])!; + /// /// In-memory test double that tracks upload/download counts and the /// high-water mark of concurrently in-flight calls, and optionally introduces an artificial From 90e1c0c48b2e0f711d15f136276a01e4a5ecb903 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 16:22:10 -0700 Subject: [PATCH 03/10] Preserve message-order failure semantics for payload fan-out RunWithBoundedConcurrencyAsync previously recorded whichever concurrent operation faulted first. This could change the existing sequential message-order contract: a later permanent PayloadStorageException could mask an earlier retriable Blob failure and incorrectly convert an OrchestratorResponse to a terminal failure. Track each fault with its operation ordinal, drain all started work, and then propagate the lowest-ordinal failure. A genuine in-flight failure continues to take precedence over concurrent cancellation, matching sequential await behavior. Add deterministic caller-level coverage where the first payload upload blocks and then raises a retriable RequestFailedException while a later payload fails permanently first. The test verifies the retriable failure escapes and no terminal response conversion occurs. Rename the existing cancellation regression to state its real-failure precedence policy. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8171f40b-65d8-4cd5-b14c-59895e8ee8e8 --- .../AzureBlobPayloadsSideCarInterceptor.cs | 71 +++++++++++++------ ...zureBlobPayloadsSideCarInterceptorTests.cs | 67 ++++++++++++++++- 2 files changed, 112 insertions(+), 26 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs index 9307ff19..40af543f 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -464,9 +464,10 @@ async Task ResolveEventPayloadsAsync(P.HistoryEvent e, CancellationToken cancell /// /// Cancellation is honored before starting any operation not already in flight. If any /// operation throws (e.g. for an oversized payload, or - /// a non-retriable ), that exception propagates the same - /// way it would have from a sequential await chain, preserving the existing first-failure - /// handling performed by callers (e.g. converting the failure into a + /// a non-retriable ), the lowest-ordinal failed operation + /// propagates after all started operations have drained. This matches a sequential await + /// chain's message-order failure semantics and takes precedence over a concurrent cancellation, + /// preserving the existing caller behavior (e.g. converting permanent failures into a /// completion). /// /// Every started operation is always drained (fully awaited) before this method returns or @@ -494,16 +495,20 @@ static async Task RunWithBoundedConcurrencyAsync(List> operations, Ca SemaphoreSlim throttle = new(MaxConcurrentPayloadOperations, MaxConcurrentPayloadOperations); List inFlight = new(operations.Count); - Exception? firstFailure = null; + object failureLock = new(); + Exception? lowestOrdinalFailure = null; + int lowestFailureOrdinal = int.MaxValue; - foreach (Func operation in operations) + for (int operationOrdinal = 0; operationOrdinal < operations.Count; operationOrdinal++) { - // Stop dispatching new Azure Storage requests once an earlier operation has failed - // permanently, or cancellation has been requested. Operations already started are - // *not* abandoned here -- they are drained below -- since they may already have side - // effects (e.g. an in-flight upload) and must not touch the semaphore, or mutate - // their target field, after this method has returned control to the caller. - if (Volatile.Read(ref firstFailure) != null || cancellation.IsCancellationRequested) + Func operation = operations[operationOrdinal]; + + // Stop dispatching new Azure Storage requests once an earlier operation has failed or + // cancellation has been requested. Operations already started are *not* abandoned here + // -- they are drained below -- since they may already have side effects (e.g. an + // in-flight upload) and must not touch the semaphore, or mutate their target field, + // after this method has returned control to the caller. + if (HasFailure() || cancellation.IsCancellationRequested) { break; } @@ -520,13 +525,13 @@ static async Task RunWithBoundedConcurrencyAsync(List> operations, Ca // release the slot back (this operation never started, so there is nothing to drain // for it) and stop dispatching -- otherwise a newly-freed slot could let a new // operation start after cancellation/failure was already observed. - if (Volatile.Read(ref firstFailure) != null || cancellation.IsCancellationRequested) + if (HasFailure() || cancellation.IsCancellationRequested) { throttle.Release(); break; } - inFlight.Add(TrackAsync(operation)); + inFlight.Add(TrackAsync(operation, operationOrdinal)); } try @@ -544,19 +549,33 @@ static async Task RunWithBoundedConcurrencyAsync(List> operations, Ca throttle.Dispose(); } - if (firstFailure != null) + Exception? failureToThrow; + lock (failureLock) { - // Rethrow the first operation failure with its original stack trace preserved, + failureToThrow = lowestOrdinalFailure; + } + + if (failureToThrow != null) + { + // Rethrow the first failure in message order with its original stack trace preserved, // matching the prior sequential await-per-field behavior (and never masked by e.g. // a later cancellation or a disposed-resource exception). - ExceptionDispatchInfo.Capture(firstFailure).Throw(); + ExceptionDispatchInfo.Capture(failureToThrow).Throw(); } // No operation failed, but cancellation may still have cut dispatch short before every // operation was started; surface that the same way a sequential await chain would have. cancellation.ThrowIfCancellationRequested(); - async Task TrackAsync(Func operation) + bool HasFailure() + { + lock (failureLock) + { + return lowestOrdinalFailure != null; + } + } + + async Task TrackAsync(Func operation, int operationOrdinal) { try { @@ -564,12 +583,18 @@ async Task TrackAsync(Func operation) } catch (Exception ex) { - // Preserve first-failure semantics (only the first exception is surfaced, - // matching the prior sequential behavior). The exception is captured -- not - // rethrown -- so Task.WhenAll above never faults, guaranteeing every operation, - // including this finally block, runs to completion before the semaphore is - // disposed. - Interlocked.CompareExchange(ref firstFailure, ex, null); + // Preserve the lowest-ordinal failure, rather than whichever concurrently running + // operation happened to finish first. The exception is captured -- not rethrown -- + // so Task.WhenAll above never faults, guaranteeing every operation, including + // this finally block, runs to completion before the semaphore is disposed. + lock (failureLock) + { + if (operationOrdinal < lowestFailureOrdinal) + { + lowestFailureOrdinal = operationOrdinal; + lowestOrdinalFailure = ex; + } + } } finally { diff --git a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs index 69d13516..103f8487 100644 --- a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs +++ b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs @@ -2,6 +2,7 @@ // Licensed under the MIT License. using System.Reflection; +using Azure; using FluentAssertions; using P = Microsoft.DurableTask.Protobuf; @@ -342,6 +343,59 @@ public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_Permanent #pragma warning restore CS0612 } + [Fact] + public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_EarlierRetriableFailureWinsOverLaterPermanentFailure() + { + // Arrange: custom status is the first operation and blocks before failing with a retriable + // Blob error. The action input is a later operation that fails immediately with a permanent + // PayloadStorageException. Sequential behavior surfaces the earlier retriable failure, so + // the response must not be converted to a terminal Failed completion. + TaskCompletionSource firstUploadStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource laterPermanentFailureObserved = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource releaseFirstUpload = new(TaskCreationOptions.RunContinuationsAsynchronously); + TrackingPayloadStore store = new(uploadAsync: async (payload, cancellationToken) => + { + if (payload == "retriable") + { + firstUploadStarted.SetResult(true); + await releaseFirstUpload.Task.WaitAsync(cancellationToken); + throw new RequestFailedException(500, "Simulated retriable Blob failure.", "ServerError", null); + } + + if (payload == "permanent") + { + laterPermanentFailureObserved.SetResult(true); + throw new PayloadStorageException("Simulated permanent payload failure."); + } + + throw new InvalidOperationException($"Unexpected payload: {payload}"); + }); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + P.OrchestratorResponse response = new() { InstanceId = "instance-1", CustomStatus = "retriable" }; + response.Actions.Add(new P.OrchestratorAction + { + Id = 1, + ScheduleTask = new P.ScheduleTaskAction { Name = "activity", Input = "permanent" }, + }); + + // Act: wait until both operations have failed or are pending, then release the earlier + // operation so its lower-ordinal retriable failure completes after the permanent one. + Task externalizeTask = ExternalizeAsync(interceptor, response, CancellationToken.None); + await firstUploadStarted.Task.WaitAsync(TimeSpan.FromSeconds(10)); + await laterPermanentFailureObserved.Task.WaitAsync(TimeSpan.FromSeconds(10)); + releaseFirstUpload.SetResult(true); + + // Assert: message order, rather than completion timing, chooses the failure. The retriable + // Blob error propagates and the permanent later failure does not replace the response with + // a terminal Failed completion. + Func act = () => externalizeTask; + (await act.Should().ThrowExactlyAsync()) + .WithMessage("Simulated retriable Blob failure."); + response.Actions.Should().ContainSingle(); + response.Actions[0].ScheduleTask.Should().NotBeNull(); + response.CustomStatus.Should().Be("retriable"); + } + [Fact] public async Task ResolveResponsePayloadsAsync_PreCancelledToken_ThrowsAndPerformsNoStoreCalls() { @@ -371,7 +425,7 @@ public async Task ResolveResponsePayloadsAsync_PreCancelledToken_ThrowsAndPerfor } [Fact] - public async Task RunWithBoundedConcurrencyAsync_CancelledWhileOperationsInFlight_DrainsBeforeReturningAndDoesNotMaskFailure() + public async Task RunWithBoundedConcurrencyAsync_InFlightRealFailureTakesPrecedenceOverCancellation() { // Arrange: 9 operations -- one more than MaxConcurrentPayloadOperations (8) -- so the // dispatch loop fills all 8 concurrency slots with genuinely in-flight operations before @@ -468,9 +522,11 @@ static Task RunWithBoundedConcurrencyAsync(IReadOnlyList> operations, /// /// In-memory test double that tracks upload/download counts and the /// high-water mark of concurrently in-flight calls, and optionally introduces an artificial - /// delay to force overlapping calls so bounded concurrency can be observed deterministically. + /// delay or a test-controlled upload operation. /// - sealed class TrackingPayloadStore(TimeSpan delay = default) : PayloadStore + sealed class TrackingPayloadStore( + TimeSpan delay = default, + Func>? uploadAsync = null) : PayloadStore { const string TokenPrefix = "test-blob://"; @@ -512,6 +568,11 @@ public override async Task UploadAsync(string payLoad, CancellationToken this.EnterCall(); try { + if (uploadAsync != null) + { + return await uploadAsync(payLoad, cancellationToken); + } + if (delay > TimeSpan.Zero) { await Task.Delay(delay, cancellationToken); From cd446f80f763a1fe1c173ded1455f43ba984c441 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 16:55:08 -0700 Subject: [PATCH 04/10] Honor pre-cancellation in single payload operation fast path RunWithBoundedConcurrencyAsync skipped cancellation handling when its one-operation fast path directly invoked the payload delegate. A pre-cancelled one-field OrchestratorResponse could therefore issue a Blob upload or convert a permanent payload failure into a terminal response instead of propagating cancellation. Check the token before invoking the fast-path operation. Add caller- level coverage for a pre-cancelled CustomStatus-only response that uses a permanent upload failure as a sentinel, verifying cancellation propagates, no upload starts, and no terminal conversion is applied. Update the existing test store to count every upload attempt, including controlled failing uploads used by cancellation coverage. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8171f40b-65d8-4cd5-b14c-59895e8ee8e8 --- .../AzureBlobPayloadsSideCarInterceptor.cs | 1 + ...zureBlobPayloadsSideCarInterceptorTests.cs | 26 ++++++++++++++++++- 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs index 40af543f..0adf4263 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -489,6 +489,7 @@ static async Task RunWithBoundedConcurrencyAsync(List> operations, Ca if (operations.Count == 1) { // Fast path: avoid semaphore/list overhead for the overwhelmingly common single-field case. + cancellation.ThrowIfCancellationRequested(); await operations[0](); return; } diff --git a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs index 103f8487..7d5d6d06 100644 --- a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs +++ b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs @@ -396,6 +396,29 @@ public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_EarlierRe response.CustomStatus.Should().Be("retriable"); } + [Fact] + public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_SingleOperationPreCancelledTokenDoesNotStartUploadOrConvertResponse() + { + // Arrange: CustomStatus is the only externalizable field, so it uses the one-operation + // fast path. The store's permanent failure is a sentinel: reaching it would cause the + // caller's permanent-failure conversion unless cancellation is observed first. + TrackingPayloadStore store = new(uploadAsync: (_, _) => throw new PayloadStorageException("Upload should not start.")); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + P.OrchestratorResponse response = new() { InstanceId = "instance-1", CustomStatus = "status" }; + using CancellationTokenSource cts = new(); + cts.Cancel(); + + // Act + Func act = () => ExternalizeAsync(interceptor, response, cts.Token); + + // Assert: pre-cancellation prevents the upload and preserves the response rather than + // converting the sentinel permanent failure into a terminal failed completion. + await act.Should().ThrowAsync(); + store.UploadCount.Should().Be(0); + response.CustomStatus.Should().Be("status"); + response.Actions.Should().BeEmpty(); + } + [Fact] public async Task ResolveResponsePayloadsAsync_PreCancelledToken_ThrowsAndPerformsNoStoreCalls() { @@ -568,6 +591,8 @@ public override async Task UploadAsync(string payLoad, CancellationToken this.EnterCall(); try { + Interlocked.Increment(ref this.uploadCount); + if (uploadAsync != null) { return await uploadAsync(payLoad, cancellationToken); @@ -578,7 +603,6 @@ public override async Task UploadAsync(string payLoad, CancellationToken await Task.Delay(delay, cancellationToken); } - Interlocked.Increment(ref this.uploadCount); string token = TokenPrefix + Guid.NewGuid().ToString("N"); lock (this.gate) { From 02a980a1f4119b0f3c37205e97c6872881debf20 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 27 Jul 2026 11:43:56 -0700 Subject: [PATCH 05/10] Serialize shared protobuf payload mutations Preserve concurrent Blob I/O while serializing assignments that mutate multiple fields of the same Google.Protobuf message. Protect OrchestrationState input/output/custom-status assignments in GetInstance and QueryInstances responses, plus CompleteOrchestration result/details externalization, with short message-level locks after the storage operation completes. Add a deterministic GetInstance regression that holds the shared OrchestrationState monitor while three controlled downloads complete. It verifies downloads overlap but the response cannot complete until the shared-message lock is released. Extend the in-memory store double with controlled download behavior and attempt counting. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8171f40b-65d8-4cd5-b14c-59895e8ee8e8 --- .../AzureBlobPayloadsSideCarInterceptor.cs | 77 ++++++++++++++--- ...zureBlobPayloadsSideCarInterceptorTests.cs | 85 ++++++++++++++++++- 2 files changed, 150 insertions(+), 12 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs index 0adf4263..36d1f24d 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -122,9 +122,30 @@ protected override async Task ResolveResponsePayloadsAsync(TResponse case P.GetInstanceResponse r when r.OrchestrationState is { } s: await RunWithBoundedConcurrencyAsync( [ - async () => s.Input = await this.MaybeResolveAsync(s.Input, cancellation), - async () => s.Output = await this.MaybeResolveAsync(s.Output, cancellation), - async () => s.CustomStatus = await this.MaybeResolveAsync(s.CustomStatus, cancellation), + async () => + { + string? input = await this.MaybeResolveAsync(s.Input, cancellation); + lock (s) + { + s.Input = input; + } + }, + async () => + { + string? output = await this.MaybeResolveAsync(s.Output, cancellation); + lock (s) + { + s.Output = output; + } + }, + async () => + { + string? customStatus = await this.MaybeResolveAsync(s.CustomStatus, cancellation); + lock (s) + { + s.CustomStatus = customStatus; + } + }, ], cancellation); break; @@ -145,9 +166,30 @@ await RunWithBoundedConcurrencyAsync( List> operations = []; foreach (P.OrchestrationState s in r.OrchestrationState) { - operations.Add(async () => s.Input = await this.MaybeResolveAsync(s.Input, cancellation)); - operations.Add(async () => s.Output = await this.MaybeResolveAsync(s.Output, cancellation)); - operations.Add(async () => s.CustomStatus = await this.MaybeResolveAsync(s.CustomStatus, cancellation)); + operations.Add(async () => + { + string? input = await this.MaybeResolveAsync(s.Input, cancellation); + lock (s) + { + s.Input = input; + } + }); + operations.Add(async () => + { + string? output = await this.MaybeResolveAsync(s.Output, cancellation); + lock (s) + { + s.Output = output; + } + }); + operations.Add(async () => + { + string? customStatus = await this.MaybeResolveAsync(s.CustomStatus, cancellation); + lock (s) + { + s.CustomStatus = customStatus; + } + }); } await RunWithBoundedConcurrencyAsync(operations, cancellation); @@ -234,8 +276,22 @@ async Task ExternalizeOrchestratorResponseAsync(P.OrchestratorResponse r, Cancel { if (a.CompleteOrchestration is { } complete) { - operations.Add(async () => complete.Result = await this.MaybeExternalizeAsync(complete.Result, cancellation)); - operations.Add(async () => complete.Details = await this.MaybeExternalizeAsync(complete.Details, cancellation)); + operations.Add(async () => + { + string? result = await this.MaybeExternalizeAsync(complete.Result, cancellation); + lock (complete) + { + complete.Result = result; + } + }); + operations.Add(async () => + { + string? details = await this.MaybeExternalizeAsync(complete.Details, cancellation); + lock (complete) + { + complete.Details = details; + } + }); } if (a.TerminateOrchestration is { } term) @@ -458,8 +514,9 @@ async Task ResolveEventPayloadsAsync(P.HistoryEvent e, CancellationToken cancell /// firing them all at once via an unbounded /// (risking Azure Storage throttling for messages with many payloads). Each delegate is /// expected to assign its result to a distinct protobuf field/element, so the relative - /// completion order between operations does not affect correctness -- only the resulting - /// number of simultaneously in-flight requests is bounded. + /// completion order between operations does not affect correctness. Blob I/O is concurrent, + /// but assignments targeting fields on the same protobuf message are serialized because + /// Google.Protobuf messages do not support concurrent mutation. /// /// /// Cancellation is honored before starting any operation not already in flight. If any diff --git a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs index 7d5d6d06..d280b0b7 100644 --- a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs +++ b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs @@ -95,6 +95,80 @@ public async Task ResolveResponsePayloadsAsync_QueryInstancesResponse_ResolvesEa } } + [Fact] + public async Task ResolveResponsePayloadsAsync_GetInstanceResponse_SerializesSharedStateMutationsAfterConcurrentDownloads() + { + // Arrange: all three fields belong to the same OrchestrationState message. The test holds + // that message's monitor while allowing the downloads to finish, so only the assignments + // (not the independent Blob I/O) should wait for the shared-message lock. + TaskCompletionSource allDownloadsStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource releaseDownloads = new(TaskCreationOptions.RunContinuationsAsynchronously); + int downloadsStarted = 0; + TrackingPayloadStore store = new(downloadAsync: async (token, cancellationToken) => + { + if (Interlocked.Increment(ref downloadsStarted) == 3) + { + allDownloadsStarted.TrySetResult(true); + } + + await releaseDownloads.Task.WaitAsync(cancellationToken); + return token switch + { + "test-blob://input" => "input", + "test-blob://output" => "output", + "test-blob://status" => "status", + _ => throw new InvalidOperationException($"Unexpected token: {token}"), + }; + }); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + P.OrchestrationState state = new() + { + Input = "test-blob://input", + Output = "test-blob://output", + CustomStatus = "test-blob://status", + }; + P.GetInstanceResponse response = new() { OrchestrationState = state }; + TaskCompletionSource stateLockAcquired = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource releaseStateLock = new(TaskCreationOptions.RunContinuationsAsynchronously); + Task holdStateLock = Task.Run(() => + { + lock (state) + { + stateLockAcquired.SetResult(true); + releaseStateLock.Task.GetAwaiter().GetResult(); + } + }); + + try + { + await stateLockAcquired.Task.WaitAsync(TimeSpan.FromSeconds(10)); + + // Act + Task resolveTask = ResolveAsync(interceptor, response, CancellationToken.None); + await allDownloadsStarted.Task.WaitAsync(TimeSpan.FromSeconds(10)); + releaseDownloads.SetResult(true); + + // Assert: Blob reads overlap, but their shared protobuf-message assignments wait for + // the synchronization boundary rather than mutating state concurrently. + await Task.Delay(100); + resolveTask.IsCompleted.Should().BeFalse("the three assignments target the same protobuf message and must wait for its lock"); + store.MaxObservedConcurrency.Should().BeGreaterThan(1, "independent Blob reads should still overlap"); + + releaseStateLock.SetResult(true); + await resolveTask.WaitAsync(TimeSpan.FromSeconds(10)); + } + finally + { + releaseDownloads.TrySetResult(true); + releaseStateLock.TrySetResult(true); + await holdStateLock; + } + + state.Input.Should().Be("input"); + state.Output.Should().Be("output"); + state.CustomStatus.Should().Be("status"); + } + [Fact] public async Task ResolveResponsePayloadsAsync_WorkItemOrchestratorRequest_ResolvesPastAndNewEventsInOrder() { @@ -549,7 +623,8 @@ static Task RunWithBoundedConcurrencyAsync(IReadOnlyList> operations, /// sealed class TrackingPayloadStore( TimeSpan delay = default, - Func>? uploadAsync = null) : PayloadStore + Func>? uploadAsync = null, + Func>? downloadAsync = null) : PayloadStore { const string TokenPrefix = "test-blob://"; @@ -622,12 +697,18 @@ public override async Task DownloadAsync(string token, CancellationToken this.EnterCall(); try { + Interlocked.Increment(ref this.downloadCount); + + if (downloadAsync != null) + { + return await downloadAsync(token, cancellationToken); + } + if (delay > TimeSpan.Zero) { await Task.Delay(delay, cancellationToken); } - Interlocked.Increment(ref this.downloadCount); lock (this.gate) { return this.tokenToValue[token]; From 83f99359bd14944ee36b9f89a722ab0352d3ae78 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 27 Jul 2026 11:49:59 -0700 Subject: [PATCH 06/10] Stabilize protobuf lock regression test Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 01dc5b7c-c742-45b7-ae8c-5cda9bcfbd2c --- ...zureBlobPayloadsSideCarInterceptorTests.cs | 21 +++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs index d280b0b7..018927fe 100644 --- a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs +++ b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs @@ -102,7 +102,7 @@ public async Task ResolveResponsePayloadsAsync_GetInstanceResponse_SerializesSha // that message's monitor while allowing the downloads to finish, so only the assignments // (not the independent Blob I/O) should wait for the shared-message lock. TaskCompletionSource allDownloadsStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); - TaskCompletionSource releaseDownloads = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource releaseDownloads = new(); int downloadsStarted = 0; TrackingPayloadStore store = new(downloadAsync: async (token, cancellationToken) => { @@ -111,7 +111,7 @@ public async Task ResolveResponsePayloadsAsync_GetInstanceResponse_SerializesSha allDownloadsStarted.TrySetResult(true); } - await releaseDownloads.Task.WaitAsync(cancellationToken); + await releaseDownloads.Task; return token switch { "test-blob://input" => "input", @@ -146,15 +146,24 @@ public async Task ResolveResponsePayloadsAsync_GetInstanceResponse_SerializesSha // Act Task resolveTask = ResolveAsync(interceptor, response, CancellationToken.None); await allDownloadsStarted.Task.WaitAsync(TimeSpan.FromSeconds(10)); - releaseDownloads.SetResult(true); + TaskCompletionSource releaseStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); + Task releaseTask = Task.Run(() => + { + releaseStarted.SetResult(true); + releaseDownloads.SetResult(true); + }); + await releaseStarted.Task.WaitAsync(TimeSpan.FromSeconds(10)); // Assert: Blob reads overlap, but their shared protobuf-message assignments wait for - // the synchronization boundary rather than mutating state concurrently. - await Task.Delay(100); - resolveTask.IsCompleted.Should().BeFalse("the three assignments target the same protobuf message and must wait for its lock"); + // the synchronization boundary rather than mutating state concurrently. Releasing the + // non-asynchronous completion source resumes the download and assignment continuations + // on releaseTask; with the shared-message lock in place, it remains blocked until that + // lock is released rather than requiring a wall-clock delay to observe the condition. + releaseTask.IsCompleted.Should().BeFalse("the three assignments target the same protobuf message and must wait for its lock"); store.MaxObservedConcurrency.Should().BeGreaterThan(1, "independent Blob reads should still overlap"); releaseStateLock.SetResult(true); + await releaseTask.WaitAsync(TimeSpan.FromSeconds(10)); await resolveTask.WaitAsync(TimeSpan.FromSeconds(10)); } finally From 16752b1eb371015314c58772cbc9c74dcda89545 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 27 Jul 2026 18:09:19 -0700 Subject: [PATCH 07/10] Harden bounded payload concurrency semantics Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 01dc5b7c-c742-45b7-ae8c-5cda9bcfbd2c --- .../AzureBlobPayloadsSideCarInterceptor.cs | 104 ++++++++-- ...zureBlobPayloadsSideCarInterceptorTests.cs | 189 ++++++++++++++++-- 2 files changed, 259 insertions(+), 34 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs index 36d1f24d..ba9fcd00 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -23,6 +23,10 @@ public sealed class AzureBlobPayloadsSideCarInterceptor(PayloadStore payloadStor // account-level throttling (see https://aka.ms/azure-storage-scalability-targets). const int MaxConcurrentPayloadOperations = 8; + Action? BeforeSharedMessageLockForTest { get; set; } + + Action? SharedMessageLockAcquiredForTest { get; set; } + /// protected override async Task ExternalizeRequestPayloadsAsync(TRequest request, CancellationToken cancellation) { @@ -125,24 +129,30 @@ await RunWithBoundedConcurrencyAsync( async () => { string? input = await this.MaybeResolveAsync(s.Input, cancellation); + this.BeforeSharedMessageLockForTest?.Invoke(); lock (s) { + this.SharedMessageLockAcquiredForTest?.Invoke(s); s.Input = input; } }, async () => { string? output = await this.MaybeResolveAsync(s.Output, cancellation); + this.BeforeSharedMessageLockForTest?.Invoke(); lock (s) { + this.SharedMessageLockAcquiredForTest?.Invoke(s); s.Output = output; } }, async () => { string? customStatus = await this.MaybeResolveAsync(s.CustomStatus, cancellation); + this.BeforeSharedMessageLockForTest?.Invoke(); lock (s) { + this.SharedMessageLockAcquiredForTest?.Invoke(s); s.CustomStatus = customStatus; } }, @@ -169,24 +179,30 @@ await RunWithBoundedConcurrencyAsync( operations.Add(async () => { string? input = await this.MaybeResolveAsync(s.Input, cancellation); + this.BeforeSharedMessageLockForTest?.Invoke(); lock (s) { + this.SharedMessageLockAcquiredForTest?.Invoke(s); s.Input = input; } }); operations.Add(async () => { string? output = await this.MaybeResolveAsync(s.Output, cancellation); + this.BeforeSharedMessageLockForTest?.Invoke(); lock (s) { + this.SharedMessageLockAcquiredForTest?.Invoke(s); s.Output = output; } }); operations.Add(async () => { string? customStatus = await this.MaybeResolveAsync(s.CustomStatus, cancellation); + this.BeforeSharedMessageLockForTest?.Invoke(); lock (s) { + this.SharedMessageLockAcquiredForTest?.Invoke(s); s.CustomStatus = customStatus; } }); @@ -279,16 +295,20 @@ async Task ExternalizeOrchestratorResponseAsync(P.OrchestratorResponse r, Cancel operations.Add(async () => { string? result = await this.MaybeExternalizeAsync(complete.Result, cancellation); + this.BeforeSharedMessageLockForTest?.Invoke(); lock (complete) { + this.SharedMessageLockAcquiredForTest?.Invoke(complete); complete.Result = result; } }); operations.Add(async () => { string? details = await this.MaybeExternalizeAsync(complete.Details, cancellation); + this.BeforeSharedMessageLockForTest?.Invoke(); lock (complete) { + this.SharedMessageLockAcquiredForTest?.Invoke(complete); complete.Details = details; } }); @@ -536,7 +556,13 @@ async Task ResolveEventPayloadsAsync(P.HistoryEvent e, CancellationToken cancell /// /// The independent operations to run. /// Cancellation token. - static async Task RunWithBoundedConcurrencyAsync(List> operations, CancellationToken cancellation) + /// Optional test callback after advisory eligibility but before dispatch claim. + /// Optional test callback after an operation's failure is recorded. + static async Task RunWithBoundedConcurrencyAsync( + List> operations, + CancellationToken cancellation, + Action? afterAdvisoryDispatchCheckForTest = null, + Action? failureRecordedForTest = null) { if (operations.Count == 0) { @@ -556,6 +582,8 @@ static async Task RunWithBoundedConcurrencyAsync(List> operations, Ca object failureLock = new(); Exception? lowestOrdinalFailure = null; int lowestFailureOrdinal = int.MaxValue; + int failureRecorded = 0; + bool cancellationPreventedDispatch = false; for (int operationOrdinal = 0; operationOrdinal < operations.Count; operationOrdinal++) { @@ -566,7 +594,7 @@ static async Task RunWithBoundedConcurrencyAsync(List> operations, Ca // -- they are drained below -- since they may already have side effects (e.g. an // in-flight upload) and must not touch the semaphore, or mutate their target field, // after this method has returned control to the caller. - if (HasFailure() || cancellation.IsCancellationRequested) + if (AdvisoryShouldStopDispatch()) { break; } @@ -578,12 +606,42 @@ static async Task RunWithBoundedConcurrencyAsync(List> operations, Ca // operation. await throttle.WaitAsync(CancellationToken.None); - // Re-check immediately after acquiring the slot: cancellation or a failure may have - // occurred while this operation was queued waiting for a slot to free up. If so, - // release the slot back (this operation never started, so there is nothing to drain - // for it) and stop dispatching -- otherwise a newly-freed slot could let a new - // operation start after cancellation/failure was already observed. - if (HasFailure() || cancellation.IsCancellationRequested) + // Avoid taking the claim lock when a failure or cancellation is already visible. This + // check is advisory: the authoritative decision remains the locked claim below. + if (AdvisoryShouldStopDispatch()) + { + throttle.Release(); + break; + } + + // A test can pause after advisory eligibility passed to deterministically exercise a + // failure racing with the authoritative dispatch claim. + afterAdvisoryDispatchCheckForTest?.Invoke(operationOrdinal); + + // Atomically claim this operation under the same lock used to record failures. The + // successful claim is the operation's dispatch linearization point: a failure recorded + // afterward does not revoke it, but no operation can claim dispatch after a failure is + // recorded. Invoke the delegate outside the lock so arbitrary operation code never + // runs while holding shared helper state. + bool dispatchClaimed; + lock (failureLock) + { + if (lowestOrdinalFailure != null) + { + dispatchClaimed = false; + } + else if (cancellation.IsCancellationRequested) + { + cancellationPreventedDispatch = true; + dispatchClaimed = false; + } + else + { + dispatchClaimed = true; + } + } + + if (!dispatchClaimed) { throttle.Release(); break; @@ -608,9 +666,11 @@ static async Task RunWithBoundedConcurrencyAsync(List> operations, Ca } Exception? failureToThrow; + bool throwCancellation; lock (failureLock) { failureToThrow = lowestOrdinalFailure; + throwCancellation = cancellationPreventedDispatch; } if (failureToThrow != null) @@ -621,16 +681,28 @@ static async Task RunWithBoundedConcurrencyAsync(List> operations, Ca ExceptionDispatchInfo.Capture(failureToThrow).Throw(); } - // No operation failed, but cancellation may still have cut dispatch short before every - // operation was started; surface that the same way a sequential await chain would have. - cancellation.ThrowIfCancellationRequested(); + if (throwCancellation) + { + // Synthesize cancellation only when it actually prevented a pending operation from + // claiming dispatch. Cancellation observed after every operation was already claimed + // must not convert otherwise successful completed work into cancellation. + cancellation.ThrowIfCancellationRequested(); + } - bool HasFailure() + bool AdvisoryShouldStopDispatch() { - lock (failureLock) + if (Volatile.Read(ref failureRecorded) != 0) + { + return true; + } + + if (cancellation.IsCancellationRequested) { - return lowestOrdinalFailure != null; + cancellationPreventedDispatch = true; + return true; } + + return false; } async Task TrackAsync(Func operation, int operationOrdinal) @@ -652,7 +724,11 @@ async Task TrackAsync(Func operation, int operationOrdinal) lowestFailureOrdinal = operationOrdinal; lowestOrdinalFailure = ex; } + + Volatile.Write(ref failureRecorded, 1); } + + failureRecordedForTest?.Invoke(operationOrdinal); } finally { diff --git a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs index 018927fe..1f391855 100644 --- a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs +++ b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs @@ -27,6 +27,12 @@ public sealed class AzureBlobPayloadsSideCarInterceptorTests static readonly MethodInfo RunWithBoundedConcurrencyMethodDefinition = typeof(AzureBlobPayloadsSideCarInterceptor) .GetMethod("RunWithBoundedConcurrencyAsync", BindingFlags.Static | BindingFlags.NonPublic)!; + static readonly PropertyInfo BeforeSharedMessageLockForTestProperty = typeof(AzureBlobPayloadsSideCarInterceptor) + .GetProperty("BeforeSharedMessageLockForTest", BindingFlags.Instance | BindingFlags.NonPublic)!; + + static readonly PropertyInfo SharedMessageLockAcquiredForTestProperty = typeof(AzureBlobPayloadsSideCarInterceptor) + .GetProperty("SharedMessageLockAcquiredForTest", BindingFlags.Instance | BindingFlags.NonPublic)!; + [Fact] public async Task ResolveResponsePayloadsAsync_HistoryChunk_ResolvesEventsInOrderWithBoundedConcurrency() { @@ -99,10 +105,10 @@ public async Task ResolveResponsePayloadsAsync_QueryInstancesResponse_ResolvesEa public async Task ResolveResponsePayloadsAsync_GetInstanceResponse_SerializesSharedStateMutationsAfterConcurrentDownloads() { // Arrange: all three fields belong to the same OrchestrationState message. The test holds - // that message's monitor while allowing the downloads to finish, so only the assignments - // (not the independent Blob I/O) should wait for the shared-message lock. + // that message's monitor while allowing the downloads to finish. A test-only callback + // synchronously signals immediately before each assignment attempts that monitor. TaskCompletionSource allDownloadsStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); - TaskCompletionSource releaseDownloads = new(); + TaskCompletionSource releaseDownloads = new(TaskCreationOptions.RunContinuationsAsynchronously); int downloadsStarted = 0; TrackingPayloadStore store = new(downloadAsync: async (token, cancellationToken) => { @@ -128,6 +134,23 @@ public async Task ResolveResponsePayloadsAsync_GetInstanceResponse_SerializesSha CustomStatus = "test-blob://status", }; P.GetInstanceResponse response = new() { OrchestrationState = state }; + TaskCompletionSource allAssignmentsReachedLock = new(TaskCreationOptions.RunContinuationsAsynchronously); + int assignmentsReachedLock = 0; + int messageLockAcquisitions = 0; + BeforeSharedMessageLockForTestProperty.SetValue(interceptor, (Action)(() => + { + if (Interlocked.Increment(ref assignmentsReachedLock) == 3) + { + allAssignmentsReachedLock.TrySetResult(true); + } + })); + SharedMessageLockAcquiredForTestProperty.SetValue(interceptor, (Action)(message => + { + message.Should().BeSameAs(state); + Monitor.IsEntered(message).Should().BeTrue(); + Interlocked.Increment(ref messageLockAcquisitions); + })); + TaskCompletionSource stateLockAcquired = new(TaskCreationOptions.RunContinuationsAsynchronously); TaskCompletionSource releaseStateLock = new(TaskCreationOptions.RunContinuationsAsynchronously); Task holdStateLock = Task.Run(() => @@ -146,24 +169,20 @@ public async Task ResolveResponsePayloadsAsync_GetInstanceResponse_SerializesSha // Act Task resolveTask = ResolveAsync(interceptor, response, CancellationToken.None); await allDownloadsStarted.Task.WaitAsync(TimeSpan.FromSeconds(10)); - TaskCompletionSource releaseStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); - Task releaseTask = Task.Run(() => - { - releaseStarted.SetResult(true); - releaseDownloads.SetResult(true); - }); - await releaseStarted.Task.WaitAsync(TimeSpan.FromSeconds(10)); - - // Assert: Blob reads overlap, but their shared protobuf-message assignments wait for - // the synchronization boundary rather than mutating state concurrently. Releasing the - // non-asynchronous completion source resumes the download and assignment continuations - // on releaseTask; with the shared-message lock in place, it remains blocked until that - // lock is released rather than requiring a wall-clock delay to observe the condition. - releaseTask.IsCompleted.Should().BeFalse("the three assignments target the same protobuf message and must wait for its lock"); + releaseDownloads.SetResult(true); + await allAssignmentsReachedLock.Task.WaitAsync(TimeSpan.FromSeconds(10)); + + // Assert: every assignment continuation has reached the lock boundary, yet none can + // mutate the shared message while its monitor is held. Blob reads still overlap. + resolveTask.IsCompleted.Should().BeFalse("all three assignments must wait for the held protobuf-message monitor"); + state.Input.Should().Be("test-blob://input"); + state.Output.Should().Be("test-blob://output"); + state.CustomStatus.Should().Be("test-blob://status"); + Volatile.Read(ref assignmentsReachedLock).Should().Be(3); + Volatile.Read(ref messageLockAcquisitions).Should().Be(0); store.MaxObservedConcurrency.Should().BeGreaterThan(1, "independent Blob reads should still overlap"); releaseStateLock.SetResult(true); - await releaseTask.WaitAsync(TimeSpan.FromSeconds(10)); await resolveTask.WaitAsync(TimeSpan.FromSeconds(10)); } finally @@ -176,6 +195,7 @@ public async Task ResolveResponsePayloadsAsync_GetInstanceResponse_SerializesSha state.Input.Should().Be("input"); state.Output.Should().Be("output"); state.CustomStatus.Should().Be("status"); + Volatile.Read(ref messageLockAcquisitions).Should().Be(3); } [Fact] @@ -614,6 +634,129 @@ public async Task RunWithBoundedConcurrencyAsync_InFlightRealFailureTakesPrecede ninthOperationDispatched.Should().BeFalse(); } + [Fact] + public async Task RunWithBoundedConcurrencyAsync_FailureRecordedBeforePendingDispatchClaim_DoesNotStartPendingOperation() + { + // Arrange: eight operations occupy the concurrency limit. Operation 0 succeeds to release + // a slot. Operation 8 passes the advisory stop check, then a test seam pauses it immediately + // before its authoritative dispatch claim. + TaskCompletionSource releaseFirstSuccess = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource releaseFailure = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource releaseRemaining = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource pendingDispatchReached = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource failureRecorded = new(TaskCreationOptions.RunContinuationsAsynchronously); + using ManualResetEventSlim continueDispatchClaim = new(initialState: false); + int operationsStarted = 0; + int pendingOperationStarted = 0; + + List> operations = + [ + async () => + { + Interlocked.Increment(ref operationsStarted); + await releaseFirstSuccess.Task; + }, + async () => + { + Interlocked.Increment(ref operationsStarted); + await releaseFailure.Task; + throw new InvalidOperationException("Lower-ordinal operation failed."); + }, + ]; + for (int i = 2; i < 8; i++) + { + operations.Add(async () => + { + Interlocked.Increment(ref operationsStarted); + await releaseRemaining.Task; + }); + } + + operations.Add(() => + { + Interlocked.Exchange(ref pendingOperationStarted, 1); + return Task.CompletedTask; + }); + + Action afterAdvisoryDispatchCheck = operationOrdinal => + { + if (operationOrdinal == 8) + { + pendingDispatchReached.TrySetResult(true); + continueDispatchClaim.Wait(); + } + }; + Action onFailureRecorded = operationOrdinal => + { + if (operationOrdinal == 1) + { + failureRecorded.TrySetResult(true); + } + }; + + Task runTask = RunWithBoundedConcurrencyAsync( + operations, + CancellationToken.None, + afterAdvisoryDispatchCheck, + onFailureRecorded); + operationsStarted.Should().Be(8); + + try + { + // Act: let operation 8 pass advisory eligibility, then record operation 1's failure + // before allowing the authoritative claim to continue. + releaseFirstSuccess.SetResult(true); + await pendingDispatchReached.Task.WaitAsync(TimeSpan.FromSeconds(10)); + releaseFailure.SetResult(true); + await failureRecorded.Task.WaitAsync(TimeSpan.FromSeconds(10)); + continueDispatchClaim.Set(); + releaseRemaining.SetResult(true); + + // Assert: the recorded failure wins the atomic claim and operation 8 never begins. + Func act = () => runTask.WaitAsync(TimeSpan.FromSeconds(10)); + (await act.Should().ThrowExactlyAsync()) + .WithMessage("Lower-ordinal operation failed."); + Volatile.Read(ref pendingOperationStarted).Should().Be(0); + } + finally + { + releaseFirstSuccess.TrySetResult(true); + releaseFailure.TrySetResult(true); + releaseRemaining.TrySetResult(true); + continueDispatchClaim.Set(); + } + } + + [Fact] + public async Task RunWithBoundedConcurrencyAsync_CancellationAfterAllOperationsComplete_ReturnsSuccessfully() + { + // Arrange: the final operation requests cancellation only after it has already claimed + // dispatch; both operations complete successfully. + using CancellationTokenSource cts = new(); + int completedOperations = 0; + List> operations = + [ + () => + { + Interlocked.Increment(ref completedOperations); + return Task.CompletedTask; + }, + () => + { + Interlocked.Increment(ref completedOperations); + cts.Cancel(); + return Task.CompletedTask; + }, + ]; + + // Act + Func act = () => RunWithBoundedConcurrencyAsync(operations, cts.Token); + + // Assert: cancellation did not prevent a dispatch, so successful work remains successful. + await act.Should().NotThrowAsync(); + completedOperations.Should().Be(2); + } + static LargePayloadStorageOptions CreateOptions() => new() { ThresholdBytes = 1 }; static Task ExternalizeAsync(AzureBlobPayloadsSideCarInterceptor interceptor, TRequest request, CancellationToken cancellation) @@ -622,8 +765,14 @@ static Task ExternalizeAsync(AzureBlobPayloadsSideCarInterceptor inter static Task ResolveAsync(AzureBlobPayloadsSideCarInterceptor interceptor, TResponse response, CancellationToken cancellation) => (Task)ResolveMethodDefinition.MakeGenericMethod(typeof(TResponse)).Invoke(interceptor, [response, cancellation])!; - static Task RunWithBoundedConcurrencyAsync(IReadOnlyList> operations, CancellationToken cancellation) - => (Task)RunWithBoundedConcurrencyMethodDefinition.Invoke(null, [operations, cancellation])!; + static Task RunWithBoundedConcurrencyAsync( + IReadOnlyList> operations, + CancellationToken cancellation, + Action? afterAdvisoryDispatchCheckForTest = null, + Action? failureRecordedForTest = null) + => (Task)RunWithBoundedConcurrencyMethodDefinition.Invoke( + null, + [operations, cancellation, afterAdvisoryDispatchCheckForTest, failureRecordedForTest])!; /// /// In-memory test double that tracks upload/download counts and the From 9554b76fb3a3a11f37f5c486913432f3130e8f6b Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 27 Jul 2026 18:29:18 -0700 Subject: [PATCH 08/10] Align reflected test helper signature Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 01dc5b7c-c742-45b7-ae8c-5cda9bcfbd2c --- .../AzureBlobPayloadsSideCarInterceptorTests.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs index 1f391855..79ce7efc 100644 --- a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs +++ b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs @@ -766,7 +766,7 @@ static Task ResolveAsync(AzureBlobPayloadsSideCarInterceptor intercep => (Task)ResolveMethodDefinition.MakeGenericMethod(typeof(TResponse)).Invoke(interceptor, [response, cancellation])!; static Task RunWithBoundedConcurrencyAsync( - IReadOnlyList> operations, + List> operations, CancellationToken cancellation, Action? afterAdvisoryDispatchCheckForTest = null, Action? failureRecordedForTest = null) From ce4b3d5d57dc1d820ad5178ec494e6821223054f Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 27 Jul 2026 18:56:33 -0700 Subject: [PATCH 09/10] Avoid protobuf lock test starvation Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 01dc5b7c-c742-45b7-ae8c-5cda9bcfbd2c --- .../AzureBlobPayloadsSideCarInterceptor.cs | 2 +- ...zureBlobPayloadsSideCarInterceptorTests.cs | 36 +++++++++++-------- 2 files changed, 22 insertions(+), 16 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs index ba9fcd00..0693d4d7 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -654,7 +654,7 @@ static async Task RunWithBoundedConcurrencyAsync( { // Wait for every started operation to finish -- regardless of outcome -- before this // method returns or throws. TrackAsync below never lets an exception fault this - // Task.WhenAll; it only records it in `firstFailure`, so draining always completes. + // Task.WhenAll; it only records it in `lowestOrdinalFailure`, so draining always completes. await Task.WhenAll(inFlight); } finally diff --git a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs index 79ce7efc..29ae43dc 100644 --- a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs +++ b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs @@ -134,14 +134,14 @@ public async Task ResolveResponsePayloadsAsync_GetInstanceResponse_SerializesSha CustomStatus = "test-blob://status", }; P.GetInstanceResponse response = new() { OrchestrationState = state }; - TaskCompletionSource allAssignmentsReachedLock = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource firstAssignmentReachedLock = new(TaskCreationOptions.RunContinuationsAsynchronously); int assignmentsReachedLock = 0; int messageLockAcquisitions = 0; BeforeSharedMessageLockForTestProperty.SetValue(interceptor, (Action)(() => { - if (Interlocked.Increment(ref assignmentsReachedLock) == 3) + if (Interlocked.Increment(ref assignmentsReachedLock) == 1) { - allAssignmentsReachedLock.TrySetResult(true); + firstAssignmentReachedLock.TrySetResult(true); } })); SharedMessageLockAcquiredForTestProperty.SetValue(interceptor, (Action)(message => @@ -153,14 +153,18 @@ public async Task ResolveResponsePayloadsAsync_GetInstanceResponse_SerializesSha TaskCompletionSource stateLockAcquired = new(TaskCreationOptions.RunContinuationsAsynchronously); TaskCompletionSource releaseStateLock = new(TaskCreationOptions.RunContinuationsAsynchronously); - Task holdStateLock = Task.Run(() => - { - lock (state) + Task holdStateLock = Task.Factory.StartNew( + () => { - stateLockAcquired.SetResult(true); - releaseStateLock.Task.GetAwaiter().GetResult(); - } - }); + lock (state) + { + stateLockAcquired.SetResult(true); + releaseStateLock.Task.GetAwaiter().GetResult(); + } + }, + CancellationToken.None, + TaskCreationOptions.LongRunning, + TaskScheduler.Default); try { @@ -170,15 +174,16 @@ public async Task ResolveResponsePayloadsAsync_GetInstanceResponse_SerializesSha Task resolveTask = ResolveAsync(interceptor, response, CancellationToken.None); await allDownloadsStarted.Task.WaitAsync(TimeSpan.FromSeconds(10)); releaseDownloads.SetResult(true); - await allAssignmentsReachedLock.Task.WaitAsync(TimeSpan.FromSeconds(10)); + await firstAssignmentReachedLock.Task.WaitAsync(TimeSpan.FromSeconds(10)); - // Assert: every assignment continuation has reached the lock boundary, yet none can - // mutate the shared message while its monitor is held. Blob reads still overlap. - resolveTask.IsCompleted.Should().BeFalse("all three assignments must wait for the held protobuf-message monitor"); + // Assert: at least one assignment has demonstrably reached the held monitor, yet none + // can mutate the shared message. Blob reads still overlap, and the inside-lock hook + // below proves all three assignments eventually acquire this exact monitor. + resolveTask.IsCompleted.Should().BeFalse("an assignment must wait for the held protobuf-message monitor"); state.Input.Should().Be("test-blob://input"); state.Output.Should().Be("test-blob://output"); state.CustomStatus.Should().Be("test-blob://status"); - Volatile.Read(ref assignmentsReachedLock).Should().Be(3); + Volatile.Read(ref assignmentsReachedLock).Should().BeGreaterThanOrEqualTo(1); Volatile.Read(ref messageLockAcquisitions).Should().Be(0); store.MaxObservedConcurrency.Should().BeGreaterThan(1, "independent Blob reads should still overlap"); @@ -195,6 +200,7 @@ public async Task ResolveResponsePayloadsAsync_GetInstanceResponse_SerializesSha state.Input.Should().Be("input"); state.Output.Should().Be("output"); state.CustomStatus.Should().Be("status"); + Volatile.Read(ref assignmentsReachedLock).Should().Be(3); Volatile.Read(ref messageLockAcquisitions).Should().Be(3); } From f8b4a90cc879ea77098b6a18e7f850c16ed075e6 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 27 Jul 2026 19:22:07 -0700 Subject: [PATCH 10/10] Drain work after dispatch callback failure Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 01dc5b7c-c742-45b7-ae8c-5cda9bcfbd2c --- .../AzureBlobPayloadsSideCarInterceptor.cs | 39 +++++++++----- ...zureBlobPayloadsSideCarInterceptorTests.cs | 53 +++++++++++++++++++ 2 files changed, 80 insertions(+), 12 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs index 0693d4d7..7a4364c2 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -616,7 +616,18 @@ static async Task RunWithBoundedConcurrencyAsync( // A test can pause after advisory eligibility passed to deterministically exercise a // failure racing with the authoritative dispatch claim. - afterAdvisoryDispatchCheckForTest?.Invoke(operationOrdinal); + try + { + afterAdvisoryDispatchCheckForTest?.Invoke(operationOrdinal); + } + catch (Exception ex) + { + // The callback runs after acquiring a slot but before claiming the operation. Route + // test failures through the normal drain path and release the unclaimed slot. + RecordFailure(ex, operationOrdinal); + throttle.Release(); + break; + } // Atomically claim this operation under the same lock used to record failures. The // successful claim is the operation's dispatch linearization point: a failure recorded @@ -705,6 +716,20 @@ bool AdvisoryShouldStopDispatch() return false; } + void RecordFailure(Exception ex, int operationOrdinal) + { + lock (failureLock) + { + if (operationOrdinal < lowestFailureOrdinal) + { + lowestFailureOrdinal = operationOrdinal; + lowestOrdinalFailure = ex; + } + + Volatile.Write(ref failureRecorded, 1); + } + } + async Task TrackAsync(Func operation, int operationOrdinal) { try @@ -717,17 +742,7 @@ async Task TrackAsync(Func operation, int operationOrdinal) // operation happened to finish first. The exception is captured -- not rethrown -- // so Task.WhenAll above never faults, guaranteeing every operation, including // this finally block, runs to completion before the semaphore is disposed. - lock (failureLock) - { - if (operationOrdinal < lowestFailureOrdinal) - { - lowestFailureOrdinal = operationOrdinal; - lowestOrdinalFailure = ex; - } - - Volatile.Write(ref failureRecorded, 1); - } - + RecordFailure(ex, operationOrdinal); failureRecordedForTest?.Invoke(operationOrdinal); } finally diff --git a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs index 29ae43dc..1a726498 100644 --- a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs +++ b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs @@ -640,6 +640,59 @@ public async Task RunWithBoundedConcurrencyAsync_InFlightRealFailureTakesPrecede ninthOperationDispatched.Should().BeFalse(); } + [Fact] + public async Task RunWithBoundedConcurrencyAsync_DispatchCallbackThrows_DrainsClaimedOperationBeforeRethrow() + { + // Arrange: operation 0 is already claimed and blocked when the test-only callback throws + // before operation 1 can claim its slot. + TaskCompletionSource releaseClaimedOperation = new(TaskCreationOptions.RunContinuationsAsynchronously); + int claimedOperationCompleted = 0; + int pendingOperationStarted = 0; + List> operations = + [ + async () => + { + await releaseClaimedOperation.Task; + Interlocked.Exchange(ref claimedOperationCompleted, 1); + }, + () => + { + Interlocked.Exchange(ref pendingOperationStarted, 1); + return Task.CompletedTask; + }, + ]; + Action afterAdvisoryDispatchCheck = operationOrdinal => + { + if (operationOrdinal == 1) + { + throw new InvalidOperationException("Test dispatch callback failed."); + } + }; + + // Act + Task runTask = RunWithBoundedConcurrencyAsync( + operations, + CancellationToken.None, + afterAdvisoryDispatchCheck); + + // Assert: the callback failure stops the pending operation but does not abandon work that + // was already claimed. The original callback exception is rethrown only after that work drains. + try + { + runTask.IsCompleted.Should().BeFalse(); + Volatile.Read(ref pendingOperationStarted).Should().Be(0); + } + finally + { + releaseClaimedOperation.TrySetResult(true); + } + + Func act = () => runTask.WaitAsync(TimeSpan.FromSeconds(10)); + (await act.Should().ThrowExactlyAsync()) + .WithMessage("Test dispatch callback failed."); + Volatile.Read(ref claimedOperationCompleted).Should().Be(1); + } + [Fact] public async Task RunWithBoundedConcurrencyAsync_FailureRecordedBeforePendingDispatchClaim_DoesNotStartPendingOperation() {