Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Bernd Verst (berndverst) merged 17 commits into from
Aug 13, 2026
Merged

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

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
4a8a0aa
Bound payload operation concurrency for fan-out messages
Jul 24, 2026
f5c0be6
Fix cancellation/disposal race in bounded-concurrency payload dispatch
Jul 24, 2026
90e1c0c
Preserve message-order failure semantics for payload fan-out
Jul 24, 2026
cd446f8
Honor pre-cancellation in single payload operation fast path
Jul 24, 2026
e8c6ceb
Merge remote-tracking branch 'origin/main' into berndverst-miniature-…
Jul 27, 2026
02a980a
Serialize shared protobuf payload mutations
Jul 27, 2026
83f9935
Stabilize protobuf lock regression test
Jul 27, 2026
16752b1
Harden bounded payload concurrency semantics
Jul 28, 2026
9554b76
Align reflected test helper signature
Jul 28, 2026
ce4b3d5
Avoid protobuf lock test starvation
Jul 28, 2026
f8b4a90
Drain work after dispatch callback failure
Jul 28, 2026
60e64b6
Merge branch 'main' into berndverst-miniature-chainsaw
berndverst Jul 31, 2026
b0f250d
Merge branch 'main' into berndverst-miniature-chainsaw
YunchuWang Jul 31, 2026
4b6923a
Address Azure Blob payload review feedback
Aug 3, 2026
f904ad9
Address remaining payload concurrency review feedback
Aug 4, 2026
bf308a0
Merge origin/main into PR #786
Aug 4, 2026
1086135
Correct concurrency test ordinals
Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
222 changes: 169 additions & 53 deletions src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,13 @@ namespace Microsoft.DurableTask;
public sealed class AzureBlobPayloadsSideCarInterceptor(PayloadStore payloadStore, LargePayloadStorageOptions options)
: PayloadInterceptor<object, object>(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;
Comment thread
berndverst marked this conversation as resolved.

/// <inheritdoc/>
protected override async Task ExternalizeRequestPayloadsAsync<TRequest>(TRequest request, CancellationToken cancellation)
{
Expand Down Expand Up @@ -112,81 +119,106 @@ protected override async Task ResolveResponsePayloadsAsync<TResponse>(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(
Comment thread
berndverst marked this conversation as resolved.
Outdated
[
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);
Comment thread
berndverst marked this conversation as resolved.
Outdated
Comment thread
berndverst marked this conversation as resolved.
Outdated
break;
case P.HistoryChunk c when c.Events != null:
foreach (P.HistoryEvent e in c.Events)
{
await this.ResolveEventPayloadsAsync(e, cancellation);
List<Func<Task>> operations = [];
foreach (P.HistoryEvent e in c.Events)
{
operations.Add(() => this.ResolveEventPayloadsAsync(e, cancellation));
}
Comment thread
berndverst marked this conversation as resolved.

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

await RunWithBoundedConcurrencyAsync(operations, cancellation);
}

break;
case P.GetEntityResponse r when r.Entity is { } em:
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<Func<Task>> 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<Func<Task>> operations = [];
Comment thread
berndverst marked this conversation as resolved.

// 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));
}
Comment thread
berndverst marked this conversation as resolved.

foreach (P.HistoryEvent? e in or.NewEvents)
{
operations.Add(() => this.ResolveEventPayloadsAsync(e, cancellation));
}
Comment thread
berndverst marked this conversation as resolved.
}
}

// 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));
}
Comment thread
berndverst marked this conversation as resolved.
}
}

await RunWithBoundedConcurrencyAsync(operations, cancellation);
}

break;
Expand All @@ -195,60 +227,64 @@ protected override async Task ResolveResponsePayloadsAsync<TResponse>(TResponse

async Task ExternalizeOrchestratorResponseAsync(P.OrchestratorResponse r, CancellationToken cancellation)
{
r.CustomStatus = await this.MaybeExternalizeAsync(r.CustomStatus, cancellation);
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)
{
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<Func<Task>> 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));
}
}
}
Expand All @@ -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<Func<Task>> 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)
Expand Down Expand Up @@ -410,6 +451,81 @@ async Task ResolveEventPayloadsAsync(P.HistoryEvent e, CancellationToken cancell
}
}

/// <summary>
/// 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 <see cref="Task.WhenAll(IEnumerable{Task})"/>
/// (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.
/// </summary>
/// <remarks>
/// Cancellation is honored before starting any operation not already in flight. If any
/// operation throws (e.g. <see cref="PayloadStorageException"/> for an oversized payload, or
/// a non-retriable <see cref="RequestFailedException"/>), 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
/// <see cref="P.TaskFailureDetails"/> completion).
/// </remarks>
/// <param name="operations">The independent operations to run.</param>
/// <param name="cancellation">Cancellation token.</param>
static async Task RunWithBoundedConcurrencyAsync(IReadOnlyList<Func<Task>> operations, CancellationToken cancellation)
{
if (operations.Count == 0)
Comment thread
berndverst marked this conversation as resolved.
{
return;
}

if (operations.Count == 1)
{
// Fast path: avoid semaphore/list overhead for the overwhelmingly common single-field case.
await operations[0]();
return;
}
Comment thread
Copilot marked this conversation as resolved.

using SemaphoreSlim throttle = new(MaxConcurrentPayloadOperations, MaxConcurrentPayloadOperations);
List<Task> inFlight = new(operations.Count);
Exception? firstFailure = null;

foreach (Func<Task> 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<Task> operation)
{
try
{
await operation();
}
catch (Exception ex)
{
Interlocked.CompareExchange(ref firstFailure, ex, null);
throw;
}
Comment thread
github-code-quality[bot] marked this conversation as resolved.
Fixed
Comment thread
berndverst marked this conversation as resolved.
finally
{
throttle.Release();
}
}
}

/// <summary>
/// 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
Expand Down
Loading
Loading