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

[berndverst]

Fixes #775 ## Summary AzureBlobPayloadsSideCarInterceptor previously externalized and resolved independent Azure Blob payloads sequentially, adding per-field latency for fan-out responses, history chunks, query results, and entity batches. This PR adds an internal bounded-concurrency helper that overlaps up to eight storage operations per message. It avoids both serial latency and the throttling risk of unbounded Task.WhenAll fan-out. ## Correctness and failure semantics - Each pending operation makes an atomic dispatch claim under the same lock used to record failures. Once a failure is recorded, no unclaimed operation can begin; already-claimed operations drain before return. - If multiple claimed operations fail, the lowest input ordinal is rethrown with its original stack, preserving the prior sequential message-order failure behavior regardless of completion order. - Cancellation stops pending dispatch but never abandons started work. Cancellation is surfaced only when it actually prevented an operation from being dispatched; cancellation observed after all successful work was claimed does not retroactively convert success into cancellation. - The semaphore is disposed only after every claimed operation, including its release path, has completed. - Permanent storage failures retain the existing activity/orchestration conversion to TaskFailureDetails or a failed completion action. ## Protobuf safety Azure Blob I/O remains concurrent, but writes to multiple fields of the same Google.Protobuf message are serialized after each storage operation completes: - OrchestrationState input, output, and custom status; - CompleteOrchestrationAction result and details. Operations that mutate distinct protobuf messages remain concurrent. Collection ordering is unchanged because closures assign to their captured field or element rather than completion order. ## Compatibility No public API, serialization, wire-format, or replay behavior change. All helper and observability seams are private implementation details and are inert when unused. ## Testing The 17 focused interceptor tests cover bounded overlap, ordering, all refactored request/response shapes, permanent failures, pre-cancellation, failure ordering, failure-versus-dispatch races, cancellation-versus-drain behavior, late cancellation after completed success, shared-message monitor ownership, and throwing pre-claim callback cleanup. Concurrency regressions use deterministic gates rather than timing assumptions. The dispatch-claim and protobuf-lock tests were red-tested against the actual pre-fix implementations. Each regression was stress-run at least 10 times; after CI exposed thread-pool starvation in the monitor test harness, its dedicated-thread form passed 20 consecutive runs. ## Validation - Focused interceptor tests: 17/17 passing. - Concurrency stress runs: 60/60 passing on the final test implementations. - Full Grpc.IntegrationTests: 159/159 passing. - AzureBlobPayloads.csproj builds for netstandard2.0, net6.0, net8.0, and net10.0. - Independent GPT-5.6 Sol max-effort adversarial review: clean.

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

[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 (1)

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

• The XML doc says assignments targeting the same protobuf message "are serialized", but RunWithBoundedConcurrencyAsync doesn't itself provide any synchronization; it just runs the delegates concurrently. The synchronization is currently implemented (selectively) by individual delegates via lock statements, so the doc should reflect that callers are responsible for synchronizing same-message mutations.

    /// (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. 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.

[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 (7)

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

• These concurrent operations lock only for assignment, but they read s.Input outside the lock before awaiting. If another operation assigns a different field on the same protobuf message while this read occurs, it can create concurrent read/write access on a Google.Protobuf message (not thread-safe). Snapshot the field value under the same lock before starting the async resolve, then assign under the lock after the await.

                        async () =>
                        {
                            string? input = await this.MaybeResolveAsync(s.Input, cancellation);
                            lock (s)
                            {

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

• Same issue as the Input resolver above: s.Output is read outside the lock before the await, which can race with other concurrent assignments to the same protobuf message. Snapshot under lock, then assign under lock.

                        async () =>
                        {
                            string? output = await this.MaybeResolveAsync(s.Output, cancellation);
                            lock (s)
                            {

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

• Same pattern: s.CustomStatus is read outside the lock prior to the await. To avoid concurrent read/write access on the same protobuf message instance, snapshot the current value under lock before the async call.

                        async () =>
                        {
                            string? customStatus = await this.MaybeResolveAsync(s.CustomStatus, cancellation);
                            lock (s)
                            {

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

• Within QueryInstancesResponse, each OrchestrationState has 3 concurrent operations. This one reads s.Input outside the lock; if another operation assigns s.Output/s.CustomStatus concurrently, this can still be concurrent access on the same protobuf message. Snapshot the source value under lock before awaiting, then assign under lock.

                        operations.Add(async () =>
                        {
                            string? input = await this.MaybeResolveAsync(s.Input, cancellation);
                            lock (s)
                            {

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

• Same issue for the Output resolver: snapshot s.Output under lock before awaiting to avoid concurrent read/write access on the same OrchestrationState protobuf message.

                        operations.Add(async () =>
                        {
                            string? output = await this.MaybeResolveAsync(s.Output, cancellation);
                            lock (s)
                            {

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

• Same issue for CustomStatus: snapshot s.CustomStatus under lock prior to awaiting to avoid concurrent access to the same protobuf message instance.

                        operations.Add(async () =>
                        {
                            string? customStatus = await this.MaybeResolveAsync(s.CustomStatus, cancellation);
                            lock (s)
                            {

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

• Same as above for complete.Details: snapshot under lock before awaiting to avoid concurrent read/write access on the same protobuf message instance.

                operations.Add(async () =>
                {
                    string? details = await this.MaybeExternalizeAsync(complete.Details, cancellation);
                    lock (complete)
                    {

[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.

[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 no new comments.