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.

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (1)

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

• The TrackAsync catch comment claims this path ensures Task.WhenAll(inFlight) “never faults”, but that’s not strictly guaranteed because other callbacks (e.g., failureRecordedForTest) could still throw and fault the task. Consider rewording the comment to describe the actual guarantee here (operation exceptions are captured so the helper can drain all started operations before disposing the semaphore), without asserting that Task.WhenAll can’t fault.

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

[Copilot]

Pull request overview

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

Suppressed comments (1)

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

• In this CompleteOrchestration block, the operations read complete.Result/complete.Details outside the lock (complete) before awaiting Blob I/O. With bounded concurrency, these per-field operations can be dispatched at different times, so a later-starting operation may read from complete while an earlier one is concurrently mutating complete under lock. Google.Protobuf messages are not safe for concurrent read/write, so snapshot these scalar values before starting bounded work and have each operation use the snapshot (writes can remain serialized under the lock).

            if (a.CompleteOrchestration is { } complete)
            {
                operations.Add(async () =>
                {
                    string? result = await this.MaybeExternalizeAsync(complete.Result, cancellation);
                    this.BeforeSharedMessageLockForTest?.Invoke();
                    lock (complete)

[YunchuWang]

Reviewed the bounded-concurrency helper and every refactored call site. The core design is careful — drain-before-return, the semaphore-disposal ordering, and lowest-ordinal failure selection are all correct, and the failure-ordering fix in 90e1c0c is the right call (a later permanent failure masking an earlier retriable one would have wrongly terminated an orchestration).

One blocking-ish concern about unconditional per-message cost on the hottest path, one evidence-based correction to a premise this PR acted on, and three consistency/scope questions. Details inline.

Minor and non-blocking: after a first failure, up to MaxConcurrentPayloadOperations - 1 already-dispatched uploads still run to completion, and the OrchestratorResponse path then calls Actions.Clear() — those blobs become orphans. Harmless for wire correctness (the previous sequential code did no post-failure I/O), but worth a line in the docs about relying on a container lifecycle policy.

[YunchuWang]

Operation lists are built unconditionally, so messages with no externalized payload now pay the full helper cost.

These three lambdas are unconditional — constructed and dispatched even when Input/Output/CustomStatus are all null or all inline values. MaybeResolveAsync returns immediately for anything that isn't a known payload token (PayloadInterceptor.cs:199-207), so in the common case this costs three closures, three TrackAsync async state machines, three Tasks, a SemaphoreSlim, a List<Task>, plus 3x WaitAsync/Release and 4x lock (failureLock) — for zero blob I/O.

The same shape appears at:

• L232 (WorkItem) — an orchestrator work item replaying N history events allocates N closures + N state machines even when no event carries a token. This is the hottest path in the worker.
• L176 (QueryInstancesResponse) — 3xN operations for N instances.
• L289 — CustomStatus is always enqueued, so an ordinary "orchestration completed with a small result" response builds a semaphore to run 3 no-op operations.

The win described in #775 is conditional on a message carrying 2+ oversized payloads, but this cost is unconditional and paid per message.

Suggestion: filter before enqueuing. It's semantically equivalent, because the calls being skipped were already no-ops:

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

With that, zero-payload messages fall into the operations.Count == 0 early return and single-payload messages into the Count == 1 fast path, so the new machinery only materializes when it can actually help.

This needs a small seam on PayloadInterceptor — both payloadStore and options.ThresholdBytes are private, so the derived interceptor can't currently test eligibility. Something like protected bool WouldResolve(string?) / protected bool WouldExternalize(string?) would be enough.

[YunchuWang]

The premise behind these locks doesn't hold for this repo's generated types.

These locks were added in response to review feedback stating that protobuf messages aren't safe for concurrent mutation "even when setting different fields". I checked the generated code, and that isn't true for these particular types.

Every setter this PR locks compiles to a single stfld with no ldfld — there is no read-modify-write of shared state. Dumped from the built Microsoft.DurableTask.Grpc.dll (Google.Protobuf 3.33.5.0) via MethodBody.GetILAsByteArray():

set_Input        [00 02 03 7D 64 02 00 04 2A]  nop; ldarg.0; ldarg.1; stfld input_;         ret
set_Output       [00 02 03 7D 67 02 00 04 2A]  nop; ldarg.0; ldarg.1; stfld output_;        ret
set_CustomStatus [00 02 03 7D 6A 02 00 04 2A]  nop; ldarg.0; ldarg.1; stfld customStatus_;  ret
set_Result       [00 02 03 7D B1 01 00 04 2A]  nop; ldarg.0; ldarg.1; stfld result_;        ret
set_Details      [00 02 03 7D B4 01 00 04 2A]  nop; ldarg.0; ldarg.1; stfld details_;       ret

The real concurrent-mutation hazard for protobuf-generated types comes from explicit presence (optional fields), where the setters share a _hasBits0 int and do _hasBits0 |= N — a genuine non-atomic read-modify-write on a field shared by unrelated properties. That doesn't apply here:

• orchestrator_service.proto uses google.protobuf.StringValue wrappers rather than optional, and there are zero optional field declarations in the protos directory.
• _hasBits occurs zero times in the generated OrchestratorService.cs.

Independent reference-type field writes are atomic and cannot disturb neighbouring fields:

• C# spec §9.6: "Reads and writes of the following data types shall be atomic: ... reference types."
• dotnet/runtime docs/design/specs/Memory-model.md: "Managed references are always aligned to their size on the given platform and accesses are atomic"; "the side-effects of memory reads and writes include only observing and changing values at specified memory locations"; and "Speculative writes are not allowed" — that last one is exactly what would otherwise permit word-tearing of an adjacent field.

Visibility of the writes is already supplied by await Task.WhenAll(inFlight) (L669), whose Interlocked-based completion is a full fence per the same document — and that is the same happens-before edge the previous sequential await implementation relied on. So the locks are removable, but only because the unconditional drain path exists; I'd keep that property in mind if the drain is ever made conditional.

If you agree, dropping lock (s) / lock (complete) also removes both *ForTest hooks (L26/L28) and the monitor-ownership test, which in turn removes the thread-pool starvation fragility that ce4b3d5 had to work around with a LongRunning thread.

If you'd rather keep the locks as future-proofing against a possible StringValue -> optional migration, could the comment say that explicitly? As written, a future reader will take the current justification at face value and may generalize it to types where the reasoning genuinely matters.

[YunchuWang]

Cancellation is observed for 1 and N operations but silently ignored for 0.

operations.Count == 0 returns before any cancellation check, while Count == 1 throws at L575 and Count >= 2 throws via L597/L700. So for the same already-cancelled token:

• HistoryChunk with 0 events -> returns normally; with 1 event -> throws OperationCanceledException.
• QueryInstancesResponse with 0 instances -> returns normally; with 1 instance -> throws.

Worth flagging that this is also a behavior change from main, where cancellation was never observed unless a payload actually crossed the threshold — MaybeExternalizeAsync / MaybeResolveAsync return early without ever touching the token. ResolveResponsePayloadsAsync_PreCancelledToken_ThrowsAndPerformsNoStoreCalls covers the throwing side but not the zero-operation case.

Either check the token before the Count == 0 return, or don't check it in the fast path — but the three arms should agree on one answer. This becomes more visible if the pre-filtering suggested in my other comment is adopted, since messages that produce N operations today would start producing 0.

[YunchuWang]

Consistency question: the HistoryState branch resolves the same three OrchestrationState fields sequentially and unlocked.

This site now resolves N history events concurrently, and for HistoryEvent.EventTypeOneofCase.HistoryState the per-event work at L519-527 mutates OrchestrationState.Input / Output / CustomStatus — the exact same three fields that GetInstanceResponse (L126) now guards with lock (s) — but sequentially and without any lock.

That's functionally fine today because those three writes are sequential within one event. But the two sites now embody different answers to "is concurrent mutation of a single OrchestrationState safe?", and a reader comparing them will get contradictory signals. Since these events reach here concurrently via both HistoryChunk (L162) and WorkItem, it would help to state which model is intended.

If the locks turn out to be unnecessary (see my comment on L133), both sites become consistent with no extra work.

[YunchuWang]

This bounds concurrency per message, not per storage account.

The accompanying comment motivates the limit in terms of storage throttling, but the SemaphoreSlim is created inside RunWithBoundedConcurrencyAsync (L580), so it's scoped to a single call. A worker processing W work items concurrently can issue up to 8xW simultaneous operations, which is still unbounded with respect to the account. If protecting the account is the actual goal, the limiter needs to live on the PayloadStore instance (or be a shared static) rather than per invocation.

Also worth considering exposing this on LargePayloadStorageOptions instead of hardcoding 8 — the right value depends heavily on payload size and account SKU, and today it can't be tuned without a rebuild.

Related follow-up, outside this diff: BlobPayloadStore.UploadAsync calls CreateIfNotExistsAsync on every upload, so bounded fan-out now also means up to 8 concurrent container-create requests per message. Caching that behind a Lazy<Task> would roughly halve the request count on the upload path and reduce exactly the throttling pressure this constant is guarding against.

[berndverst]

Quantitative performance impact

The live implementation at b0f250d uses a per-message concurrency bound of C = 8. The numbers below combine an exact latency model with controlled fake-store sanity checks; they are not Azure production averages.

For N independent payload operations of equal duration L:

• Previous serial path: Tserial = N * L
• New bounded path, ignoring scheduler overhead: T8 = ceil(N / 8) * L
• Ideal speedup is capped at min(N, 8)

Average-style representative and best cases

Scenario	20 ms/op, serial -> C=8	50 ms/op, serial -> C=8	100 ms/op, serial -> C=8	Ideal latency / throughput change
3 payload fields	60 -> 20 ms	150 -> 50 ms	300 -> 100 ms	3x, -66.7% latency
Representative: 8 fields	160 -> 20 ms	400 -> 50 ms	800 -> 100 ms	8x, -87.5% latency
Best sustained: 32 fields	640 -> 80 ms	1,600 -> 200 ms	3,200 -> 400 ms	8x, -87.5% latency

At the representative 50 ms point, the payload stage rises from 20 to 160 operations/s. For an eight-field message whose latency is dominated by these I/Os, the ideal message rate rises from 2.5 to 20 messages/s. Real gains will be lower when serialization, gRPC, SDK overhead, service contention, or unequal operation durations matter.

With one 100 ms tail operation and all other operations at 20 ms, modeled speedup falls to:

Fields	Serial	Bounded	Speedup
3	140 ms	100 ms	1.4x
8	240 ms	100 ms	2.4x
32	720 ms	100-160 ms depending on ordinal	4.5-7.2x

Worst, adverse, and no-gain cases

• N=0/1: no concurrency gain. The single-operation fast path avoids the semaphore, though callers still construct the operation/delegate list.
• If concurrency inflates every operation by factor q, break-even is q = N / ceil(N/8): 3x for N=3, and 8x for N=8 or N=32.
• Illustrative contention case, not a predicted Azure result: for 32 fields at a serial 50 ms/op, if all four bounded waves slow to 500 ms/op (q=10), latency changes from 1.6 to 2.0 seconds (+25%) and payload throughput from 20 to 16 operations/s (-20%). There is no finite worst case without bounding service-side contention.
• Failure latency can increase because started peers drain before return. If the first operation fails at 20 ms while seven already-claimed, non-cancellable peers take 100 ms, return moves from 20 to 100 ms (5x as long). This buys the guarantee that no started operation mutates protobuf state after return.

CPU and memory

32 equal-size operations	Serial	Bounded (C=8)	Unbounded
Simultaneously active operations	1	8	32
Active payload-buffer storage for 1 MiB buffers	~1 MiB	~8 MiB	~32 MiB

The buffer row covers active payload byte[] storage only, not source strings, SDK/compression buffers, final retained payloads, or process peak working set. Bounded concurrency is therefore approximately 75% below unbounded transient buffer pressure, but can be up to 8x serial pressure.

Total encode/compress/decode CPU work is unchanged, plus O(N) task, semaphore, claim-lock, and WhenAll bookkeeping. Instantaneous CPU pressure can rise from one operation to as many as eight, which exchanges higher short-term resource use for lower elapsed latency. The protobuf monitor locks cover only short post-I/O assignments, not network operations.

Controlled sanity check

Using the exact implementation with the existing fake store and Task.Delay(50):

• 20 downloads completed in 222 ms (ideal bounded delay 150 ms; serial delay 1,000 ms)
• 17 uploads completed in 204 ms (ideal bounded delay 150 ms; serial delay 850 ms)

These checks validate overlap and wave shape. Their 36-48% overhead above the delay-only ideal includes JIT, test setup, GUID/dictionary work, and assertions, so they should not be interpreted as Azure latency benchmarks.