Skip to content

Remove DTS Autochunking #839

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
sophiatev merged 4 commits into from
Nov 5, 2026
Merged

Remove DTS Autochunking #839

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ed77f2c
removed autochunking
Nov 5, 2026
daf4195
fixed line endings
Nov 5, 2026
96af49f
Merge branch 'main' into stevosyan/remove-auto-chunking
sophiatev Nov 5, 2026
250a7a1
addressing PR comments
Nov 5, 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
35 changes: 3 additions & 32 deletions samples/LargePayloadConsoleApp/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,9 +32,8 @@
// Output: Status: Failed / PASS / Error: Payload size 16384 KB exceeds...
//
// [Scenario 4] 13MB activity input -> echo -> orchestration output
// Validates: ValidateActionsSize bypass (13MB > 3.9MB chunk limit) + ScheduleTask.Input externalization
// Validates: ScheduleTask.Input externalization
// + ActivityResponse.Result externalization + CompleteOrchestration.Result externalization
// + isChunkedMode: single oversized action sent as non-chunked (no ChunkIndex)
// Expected: Status: Completed, Output length: 13631488, PASS
// Output: Status: Completed / Output length: 13631488 / PASS: 13MB activity output -> orch output externalized correctly
//
Expand All @@ -44,7 +43,7 @@
// Output: Status: Completed / PASS
//
// [Scenario 6] 13MB sub-orchestration input -> child returns it
// Validates: CreateSubOrchestration.Input externalization (>3.9MB action in orchestration completion)
// Validates: CreateSubOrchestration.Input externalization
// Expected: Status: Completed, PASS
// Output: Status: Completed / PASS
//
Expand All @@ -58,12 +57,6 @@
// Expected: Status: Completed, PASS
// Output: Status: Completed / PASS
//
// [Scenario 9] 3x 13MB activity inputs (real multi-chunk LP completion)
// Validates: Multiple oversized ScheduleTask actions → real chunked completion with isChunkedMode=true
// + all 3 activities execute and return correct data
// Expected: Status: Completed, PASS: total=40894464
// Output: Status: Completed / PASS: total=40894464
//
// =================================================================================================

using Microsoft.DurableTask.Client;
Expand Down Expand Up @@ -180,7 +173,7 @@ await ctx.Entities.CallEntityAsync(
return Task.FromResult(new string('P', 16 * 1024 * 1024));
});

// 13MB scenarios: validates that ValidateActionsSize does not block externalization
// 13MB payload examples
tasks.AddOrchestratorFunc<object?, string>("LargeActivityIO", async (ctx, _) =>
{
// Send 13MB as activity input, activity echoes it back, orchestration returns it
Expand Down Expand Up @@ -214,16 +207,6 @@ await ctx.Entities.CallEntityAsync(
ctx.SetCustomStatus(new string('C', 13 * 1024 * 1024));
return Task.FromResult("done");
});

// Scenario 9: 3x 13MB activity inputs (chunked orchestration completion)
tasks.AddOrchestratorFunc<object?, int>("ThreeLargeActivities", async (ctx, _) =>
{
var t1 = ctx.CallActivityAsync<string>("Echo13MB", new string('A', 13 * 1024 * 1024));
var t2 = ctx.CallActivityAsync<string>("Echo13MB", new string('B', 13 * 1024 * 1024));
var t3 = ctx.CallActivityAsync<string>("Echo13MB", new string('C', 13 * 1024 * 1024));
string[] results = await Task.WhenAll(t1, t2, t3);
return results.Sum(r => r.Length);
});
});

// Use shared store (no duplication of options)
Expand Down Expand Up @@ -326,7 +309,6 @@ await ctx.Entities.CallEntityAsync(
if (orchOvfResult.FailureDetails != null) Console.WriteLine(" Error: " + orchOvfResult.FailureDetails.ErrorMessage);

// Scenario 4: 13MB activity output + orchestration output with LP enabled
// Tests that ValidateActionsSize bypass allows 13MB through the interceptor
Console.WriteLine();
Console.WriteLine("[Scenario 4] 13MB activity output -> orchestration output (round-trip)");
string largeIOId = await client.ScheduleNewOrchestrationInstanceAsync("LargeActivityIO");
Expand Down Expand Up @@ -387,17 +369,6 @@ await ctx.Entities.CallEntityAsync(
Console.WriteLine(csResult.RuntimeStatus == OrchestrationRuntimeStatus.Completed && customStatus?.Length == 13 * 1024 * 1024
? "PASS" : "FAIL: custom status length=" + (customStatus?.Length ?? 0));

// Scenario 9: 3x 13MB activity inputs (chunked orchestration completion with LP)
Console.WriteLine();
Console.WriteLine("[Scenario 9] 3x 13MB activity inputs (chunked orch complete)");
string threeId = await client.ScheduleNewOrchestrationInstanceAsync("ThreeLargeActivities");
OrchestrationMetadata threeResult = await client.WaitForInstanceCompletionAsync(threeId, getInputsAndOutputs: true, cts.Token);
Console.WriteLine(" Status: " + threeResult.RuntimeStatus);
int totalLen = threeResult.ReadOutputAs<int>();
Console.WriteLine(threeResult.RuntimeStatus == OrchestrationRuntimeStatus.Completed && totalLen == 3 * 13 * 1024 * 1024
? "PASS: total=" + totalLen : "FAIL: total=" + totalLen);
if (threeResult.FailureDetails != null) Console.WriteLine(" Error: " + threeResult.FailureDetails.ErrorMessage);

Console.WriteLine();
Console.WriteLine("=== ALL SCENARIOS DONE ===");

Expand Down
4 changes: 0 additions & 4 deletions src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,10 +50,6 @@ internal async Task PrepareOrchestratorResponseAsync(P.OrchestratorResponse resp
response.Actions.Clear();
response.Actions.AddRange(unlocks);
response.CustomStatus = null;
#pragma warning disable CS0612 // isPartial/chunkIndex are deprecated but still required for chunked response wire compatibility.
response.IsPartial = false;
response.ChunkIndex = null;
#pragma warning restore CS0612
response.Actions.Add(new P.OrchestratorAction
{
CompleteOrchestration = new P.CompleteOrchestrationAction
Expand Down
4 changes: 1 addition & 3 deletions src/InProcessTestHost/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,9 +98,7 @@ leaving execution pending or repeatedly requeueing unsupported work. Invalid tok
mismatched identities still leave valid deliveries untouched.

Direct callers must send all actions in one response, with `isPartial` false and `chunkIndex`
omitted, within the configured SDK chunk-size and gRPC message-size limits. The SDK's
existing oversized-response chunking path is unchanged: if it sends chunks to this test
host, the orchestration fails terminally; chunking is not supported here.
omitted, within the configured gRPC message-size limits.

Replay history streaming is separate and remains supported. A full response releases the
delivery's temporary replay snapshot, while an already captured history reader can still finish
Expand Down
215 changes: 52 additions & 163 deletions src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@
using DurableTask.Core.Entities;
using DurableTask.Core.Entities.OperationFormat;
using DurableTask.Core.History;
using Google.Protobuf;
using Microsoft.DurableTask.Abstractions;
using Microsoft.DurableTask.Entities;
using Microsoft.DurableTask.Tracing;
Expand All @@ -30,6 +29,9 @@ sealed partial class GrpcDurableTaskWorker
{
class Processor
{
// Leave room below the default 4 MiB gRPC receive limit.
const int MaxInlineOrchestratorResponseSizeInBytes = 4_089_446;

static readonly Google.Protobuf.WellKnownTypes.Empty EmptyMessage = new();

readonly GrpcDurableTaskWorker worker;
Expand Down Expand Up @@ -691,13 +693,15 @@ async Task OnRunOrchestratorAsync(

if (isInitialRewind)
{
await this.CompleteOrchestratorTaskWithChunkingAsync(
RewindOrchestrationHandler.CreateResponse(
request,
pastEvents,
completionToken,
traceActivity),
this.worker.grpcOptions.CompleteOrchestrationWorkItemChunkSizeInBytes,
P.OrchestratorResponse rewindResponse = RewindOrchestrationHandler.CreateResponse(
request,
pastEvents,
completionToken,
traceActivity);
await this.PrepareOrchestratorResponseAsync(rewindResponse, null, cancellationToken);
await this.ExecuteWithRetryAsync(
async () => await this.client.CompleteOrchestratorTaskAsync(rewindResponse, cancellationToken: cancellationToken),
nameof(this.client.CompleteOrchestratorTaskAsync),
cancellationToken);
return;
}
Expand Down Expand Up @@ -929,19 +933,11 @@ await this.ExecuteWithRetryAsync(
};
}

if (this.internalOptions.PrepareOrchestratorResponseAsync is { } prepareResponse)
{
await prepareResponse(response, cancellationToken);
}

var completeOrchestrationAction = response.Actions.FirstOrDefault(
a => a.CompleteOrchestration is not null);
P.OrchestratorAction? completeOrchestrationAction = await this.PrepareOrchestratorResponseAsync(
response, entityConversionState, cancellationToken);

if (completeOrchestrationAction is not null)
{
// Cover direct or newly terminal responses; converter cleanup already drains its obligations.
entityConversionState?.InsertMissingUnlocks(response, completeOrchestrationAction);

if (completeOrchestrationAction.CompleteOrchestration.OrchestrationStatus == P.OrchestrationStatus.Failed)
{
traceActivity?.SetStatus(
Expand All @@ -965,9 +961,9 @@ await this.ExecuteWithRetryAsync(
GetActionsListForLogging(response.Actions));
}

await this.CompleteOrchestratorTaskWithChunkingAsync(
response,
this.worker.grpcOptions.CompleteOrchestrationWorkItemChunkSizeInBytes,
await this.ExecuteWithRetryAsync(
async () => await this.client.CompleteOrchestratorTaskAsync(response, cancellationToken: cancellationToken),
nameof(this.client.CompleteOrchestratorTaskAsync),
Comment thread
sophiatev marked this conversation as resolved.
Resolved
cancellationToken);
}

Expand Down Expand Up @@ -1170,165 +1166,58 @@ await this.ExecuteWithRetryAsync(
cancellation);
}

/// <summary>
/// Completes an orchestration task with automatic chunking if the response exceeds the maximum size.
/// </summary>
/// <param name="response">The orchestrator response to send.</param>
/// <param name="maxChunkBytes">The maximum size in bytes for each chunk.</param>
/// <param name="cancellationToken">The cancellation token.</param>
async Task CompleteOrchestratorTaskWithChunkingAsync(
async Task<P.OrchestratorAction?> PrepareOrchestratorResponseAsync(
P.OrchestratorResponse response,
int maxChunkBytes,
ProtoUtils.EntityConversionState? entityConversionState,
CancellationToken cancellationToken)
{
// Validate that no single action exceeds the maximum chunk size
static P.TaskFailureDetails? ValidateActionsSize(IEnumerable<P.OrchestratorAction> actions, int maxChunkBytes)
if (this.internalOptions.PrepareOrchestratorResponseAsync is { } prepareResponse)
{
foreach (P.OrchestratorAction action in actions)
{
int actionSize = action.CalculateSize();
if (actionSize > maxChunkBytes)
{
// TODO: large payload doc is not available yet on aka.ms, add doc link to below error message
string errorMessage = $"A single orchestrator action of type {action.OrchestratorActionTypeCase} with id {action.Id} " +
$"exceeds the {maxChunkBytes / 1024.0 / 1024.0:F2}MB limit: {actionSize / 1024.0 / 1024.0:F2}MB. " +
"Enable large-payload externalization to Azure Blob Storage to support oversized actions.";
return new P.TaskFailureDetails
{
ErrorType = typeof(InvalidOperationException).FullName,
ErrorMessage = errorMessage,
IsNonRetriable = true,
};
}
}
await prepareResponse(response, cancellationToken);
}

return null;
P.OrchestratorAction? completion = response.Actions.FirstOrDefault(
action => action.CompleteOrchestration is not null);
if (completion is not null)
{
entityConversionState?.InsertMissingUnlocks(response, completion);
}

P.TaskFailureDetails? validationFailure = this.worker.grpcOptions.Capabilities.Contains(P.WorkerCapability.LargePayloads)
? null
: ValidateActionsSize(response.Actions, maxChunkBytes);
if (validationFailure != null)
// Measure the full response after payload uploads and entity cleanup.
int responseSize = response.CalculateSize();
if (responseSize > MaxInlineOrchestratorResponseSizeInBytes)
Comment thread
sophiatev marked this conversation as resolved.
Resolved
{
// Complete the orchestration with a failed status and failure details
P.OrchestratorResponse failureResponse = new()
// Keep releases that were already added or sent by the orchestrator.
P.OrchestratorAction[] unlocks = [.. response.Actions.Where(action =>
action.SendEntityMessage?.EntityUnlockSent is not null
|| (action.SendEvent is { Name: "release", Instance: { } instance }
&& DTCore.Common.Entities.IsEntityInstance(instance.InstanceId))),];
response.Actions.Clear();
response.Actions.AddRange(unlocks);
response.CustomStatus = null;
completion = new P.OrchestratorAction
{
InstanceId = response.InstanceId,
CompletionToken = response.CompletionToken,
OrchestrationTraceContext = response.OrchestrationTraceContext,
Actions =
Id = completion?.Id ?? 0,
CompleteOrchestration = new P.CompleteOrchestrationAction
{
new P.OrchestratorAction
OrchestrationStatus = P.OrchestrationStatus.Failed,
FailureDetails = new P.TaskFailureDetails
{
CompleteOrchestration = new P.CompleteOrchestrationAction
{
OrchestrationStatus = P.OrchestrationStatus.Failed,
FailureDetails = validationFailure,
},
ErrorType = typeof(InvalidOperationException).FullName,
ErrorMessage = $"Orchestrator response size of {responseSize / 1024.0 / 1024.0:F2} MiB " +
$"exceeds the {MaxInlineOrchestratorResponseSizeInBytes / 1024.0 / 1024.0:F2} MiB inline limit. " +
"Enable large-payload externalization to Azure Blob Storage or reduce the response size.",
IsNonRetriable = true,
},
},
};
response.Actions.Add(completion);

await this.ExecuteWithRetryAsync(
async () => await this.client.CompleteOrchestratorTaskAsync(failureResponse, cancellationToken: cancellationToken),
nameof(this.client.CompleteOrchestratorTaskAsync),
cancellationToken);
return;
}

// Helper to add an action to the current chunk if it fits
static bool TryAddAction(
Google.Protobuf.Collections.RepeatedField<P.OrchestratorAction> dest,
P.OrchestratorAction action,
ref int currentSize,
int maxChunkBytes)
{
int actionSize = action.CalculateSize();
if (currentSize + actionSize > maxChunkBytes && currentSize > 0)
{
return false;
}

dest.Add(action);
currentSize += actionSize;
return true;
}

// Check if the entire response fits in one chunk
int totalSize = response.CalculateSize();
if (totalSize <= maxChunkBytes)
{
// Response fits in one chunk, send it directly (isPartial defaults to false)
await this.ExecuteWithRetryAsync(
async () => await this.client.CompleteOrchestratorTaskAsync(response, cancellationToken: cancellationToken),
nameof(this.client.CompleteOrchestratorTaskAsync),
cancellationToken);
return;
// A newly failed response must also release any locks still held.
entityConversionState?.InsertMissingUnlocks(response, completion);
}

// Response is too large, split into multiple chunks
int actionsCompletedSoFar = 0, chunkIndex = 0;
List<P.OrchestratorAction> allActions = response.Actions.ToList();
bool isPartial = true;
bool isChunkedMode = false;

while (isPartial)
{
P.OrchestratorResponse chunkedResponse = new()
{
InstanceId = response.InstanceId,
CustomStatus = response.CustomStatus,
CompletionToken = response.CompletionToken,
RequiresHistory = response.RequiresHistory,
NumEventsProcessed = 0,
};

int chunkPayloadSize = 0;

// Fill the chunk with actions until we reach the size limit
while (actionsCompletedSoFar < allActions.Count &&
TryAddAction(chunkedResponse.Actions, allActions[actionsCompletedSoFar], ref chunkPayloadSize, maxChunkBytes))
{
actionsCompletedSoFar++;
}

// Determine if this is a partial chunk (more actions remaining)
#pragma warning disable CS0612 // isPartial/chunkIndex are deprecated but still required for chunked response wire compatibility.
isPartial = actionsCompletedSoFar < allActions.Count;
chunkedResponse.IsPartial = isPartial;

// Only activate chunked mode when we actually need multiple chunks.
// A single oversized action that fits in one chunk (via TryAddAction allowing
// the first item in an empty chunk) should be sent as non-chunked to avoid
// backend issues with ChunkIndex=0 + IsPartial=false.
if (isPartial)
{
isChunkedMode = true;
}

if (isChunkedMode)
{
chunkedResponse.ChunkIndex = chunkIndex;
}
#pragma warning restore CS0612

if (chunkIndex == 0)
{
// The first chunk preserves the original response's NumEventsProcessed value (null)
// When this is set to null, backend by default handles all the messages in the workitem.
// For subsequent chunks, we set it to 0 since all messages are already handled in first chunk.
chunkedResponse.NumEventsProcessed = null;
chunkedResponse.OrchestrationTraceContext = response.OrchestrationTraceContext;
}

chunkIndex++;

// Send the chunk
await this.ExecuteWithRetryAsync(
async () => await this.client.CompleteOrchestratorTaskAsync(chunkedResponse, cancellationToken: cancellationToken),
nameof(this.client.CompleteOrchestratorTaskAsync),
cancellationToken);
}
return completion;
}

async Task ExecuteWithRetryAsync(
Expand Down
Loading
Loading