Perf: bound Azure Blob payload operation concurrency for fan-out messages

[berndverst]

Summary

AzureBlobPayloadsSideCarInterceptor previously externalized/resolved each Azure Blob payload sequentially (one await per field), which adds serial latency proportional to the number of independent payloads in a message — for example, a fan-out orchestrator response with many scheduled-task actions, a history chunk with many events, or an entity batch with many operations.

This PR introduces a safe, internal-only bounded-concurrency helper, RunWithBoundedConcurrencyAsync, and uses it to run each message's independent payload operations with up to 8 simultaneous Azure Blob requests instead of either:

• awaiting them one at a time (serial latency), or
• an unbounded Task.WhenAll (risking Azure Storage account-level throttling for messages with many payloads).

What changed

• Added MaxConcurrentPayloadOperations = 8 and a RunWithBoundedConcurrencyAsync helper (uses a SemaphoreSlim to bound in-flight requests).
• Refactored ResolveResponsePayloadsAsync (for GetInstanceResponse, HistoryChunk, QueryInstancesResponse, QueryEntitiesResponse, WorkItem), ExternalizeOrchestratorResponseAsync, ExternalizeEntityBatchResultAsync, and ExternalizeEntityBatchRequestAsync to build a list of independent operation closures (each assigning to its own distinct protobuf field/element) and run them through the new helper.

Preserved behavior

• Protobuf field assignment correctness & ordering: each closure captures and assigns to a distinct field/collection element; collection order is unaffected since ordering is index-based, not completion-order-based.
• Cancellation: cancellation.ThrowIfCancellationRequested() is checked before starting any not-yet-started operation.
• First-failure semantics: once any operation fails, no new operations are started (already in-flight ones are allowed to drain since they may have side effects), and the first exception propagates via Task.WhenAll — matching the prior sequential-await behavior that callers depend on (e.g. conversion to TaskFailureDetails/failed-completion actions on permanent storage failures).
• Azure Storage throttling safety: concurrency is capped at 8 simultaneous requests per message instead of unbounded.
• Replay determinism: no orchestrator-visible non-determinism introduced; this is sidecar-side gRPC interceptor code, not orchestrator logic.
• Public API: no public API changes — AzureBlobPayloadsSideCarInterceptor is sealed and all changes are internal implementation details.

Testing

Added test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs with 10 focused unit tests (via reflection against the interceptor's protected methods and an in-memory PayloadStore test double) covering:

• Ordering correctness for resolve (HistoryChunk, QueryInstancesResponse, WorkItem/OrchestratorRequest, WorkItem/EntityRequest) and externalize (OrchestratorResponse, EntityBatchRequest, EntityBatchResult) paths.
• Bounded concurrency is observed (> 1 and <= 8 simultaneous store calls) under artificial delay.
• Cancellation before any store call when a pre-cancelled token is passed.
• Permanent-failure conversion to TaskFailureDetails (activity response) and to a single failed CompleteOrchestration action (orchestrator response), with no store upload for the oversized payload.

Ran targeted validation:

dotnet build test\Grpc.IntegrationTests\Grpc.IntegrationTests.csproj
dotnet test test\Grpc.IntegrationTests\Grpc.IntegrationTests.csproj --filter "FullyQualifiedName~AzureBlobPayloadsSideCarInterceptorTests"

All 10 new tests pass, and the full Grpc.IntegrationTests suite (152 tests) passes.

Also verified dotnet build for AzureBlobPayloads.csproj succeeds across all 4 target frameworks (netstandard2.0;net6.0;net8.0;net10.0).

Fixes #775

[Copilot]

Pull request overview

This PR improves Azure Blob large-payload handling performance in the gRPC sidecar interceptor by replacing per-field sequential awaits with a bounded-concurrency execution model, reducing additive latency in fan-out messages while avoiding unbounded parallelism that could trigger Azure Storage throttling.

Changes:

• Introduced a bounded-concurrency helper (RunWithBoundedConcurrencyAsync) capped at 8 concurrent payload operations per message.
• Refactored response resolution and request externalization paths to batch independent payload operations and execute them through the helper.
• Added focused tests validating ordering, bounded concurrency, cancellation behavior, and permanent-failure conversion behavior.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs	Refactors payload resolve/externalize logic to run independent store ops with bounded concurrency via a new helper.
test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs	Adds tests for ordering correctness, concurrency bounds, cancellation, and permanent-failure handling for the refactored interceptor behavior.

Comments suppressed due to low confidence (1)

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs:504

• After awaiting throttle.WaitAsync, the loop doesn't re-check firstFailure before starting the next operation. If another in-flight operation fails while this iteration is blocked in WaitAsync, this iteration can still enqueue a new operation (violating the intended “stop starting new operations after first failure” behavior). Also, if multiple operations fault, relying on Task.WhenAll to pick which exception to throw can drift from the intended first-failure semantics.

            await throttle.WaitAsync(cancellation);
            inFlight.Add(TrackAsync(operation));

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (10)

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs:150

• These queued operations use expression-bodied async lambdas that return Task<string?> (assignment expression), which is incompatible with List<Func> and will fail to compile. Wrap the assignment in a statement-bodied async lambda.

                        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));

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs:165

• This operation is an expression-bodied async lambda that returns Task<string?> due to the assignment expression, which won’t compile when added to List<Func>. Use a statement-bodied async lambda.

                        operations.Add(async () => em.SerializedState = await this.MaybeResolveAsync(em.SerializedState, cancellation));

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs:179

• This queued operation is an expression-bodied async lambda that returns Task<string?> (assignment expression), which won’t compile when targeting Func. Wrap the assignment in braces so the lambda returns Task.

                        operations.Add(async () => ar.Input = await this.MaybeResolveAsync(ar.Input, cancellation));

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs:203

• The queued operations that assign EntityState and OperationRequest.Input are expression-bodied async lambdas, so they return Task<string?> and won’t compile when stored as Func. Use statement-bodied async lambdas for these assignments.

                        operations.Add(async () => er1.EntityState = await this.MaybeResolveAsync(er1.EntityState, cancellation));
                        if (er1.Operations != null)
                        {
                            foreach (P.OperationRequest op in er1.Operations)
                            {

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs:212

• This queued operation is an expression-bodied async lambda that returns Task<string?> (assignment expression), which is incompatible with Func. Wrap the assignment in a statement-bodied async lambda.

                        operations.Add(async () => er2.EntityState = await this.MaybeResolveAsync(er2.EntityState, cancellation));

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs:235

• The operations list is populated with expression-bodied async lambdas whose bodies are assignment expressions, so they return Task<string?> and won’t compile as Func. Convert these to statement-bodied async lambdas (wrap in { ...; }).

        List<Func<Task>> operations = [async () => r.CustomStatus = await this.MaybeExternalizeAsync(r.CustomStatus, cancellation)];

        foreach (P.OrchestratorAction a in r.Actions)
        {
            if (a.CompleteOrchestration is { } complete)

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs:260

• These operations are also expression-bodied async lambdas returning Task<string?> due to assignment expressions, which won’t compile when stored as Func. Wrap each assignment in a statement-bodied async lambda.

            if (a.SendEvent is { } sendEvt)
            {
                operations.Add(async () => sendEvt.Data = await this.MaybeExternalizeAsync(sendEvt.Data, cancellation));
            }

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs:284

• These operations use expression-bodied async lambdas that return Task<string?> (assignment expression), which is incompatible with List<Func> and will not compile. Use statement-bodied async lambdas for these assignments.

        List<Func<Task>> operations = [async () => r.EntityState = await this.MaybeExternalizeAsync(r.EntityState, cancellation)];

        if (r.Results != null)
        {
            foreach (P.OperationResult result in r.Results)

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs:301

• The queued operations for SendSignal/Input and StartNewOrchestration/Input are expression-bodied async lambdas returning Task<string?> (assignment expression), which won’t compile as Func. Wrap each assignment in a statement-bodied async lambda.

                if (action.SendSignal is { } sendSig)
                {
                    operations.Add(async () => sendSig.Input = await this.MaybeExternalizeAsync(sendSig.Input, cancellation));
                }

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs:318

• These operations use expression-bodied async lambdas (assignment expressions), which return Task<string?> and won’t compile when stored as Func. Wrap the assignments in statement-bodied async lambdas.

        List<Func<Task>> operations = [async () => r.EntityState = await this.MaybeExternalizeAsync(r.EntityState, cancellation)];

        if (r.Operations != null)
        {
            foreach (P.OperationRequest op in r.Operations)

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (2)

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs:151

• This builds 3 independent operations per OrchestrationState instance, so the bounded-concurrency runner may set Input/Output/CustomStatus on the same P.OrchestrationState concurrently. Protobuf messages aren't safe for concurrent mutation; a safer approach is to keep concurrency across instances but resolve each instance's fields sequentially within a single operation.

                    List<Func<Task>> 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));
                    }

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs:239

• Within a single CompleteOrchestration action, Result and Details are externalized via two separate bounded-concurrency operations. That can concurrently mutate the same protobuf message (complete), which is not thread-safe for concurrent writes. Combine these into one operation so each message instance is only mutated by one task at a time.

            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));
            }