From bb4aefec7219615fd995fc6b840cf45e1f7b09ed Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 15:07:13 -0700 Subject: [PATCH 1/5] Perf: avoid repeated protobuf sizing during orchestration response chunking CompleteOrchestratorTaskWithChunkingAsync previously called CalculateSize() on every action up to three times: once during oversized-action validation, once implicitly via response.CalculateSize() for the whole-response check, and once more per action inside TryAddAction while packing chunks. Now the whole response is sized exactly once up front. If it fits within maxChunkBytes, it is sent directly with no per-action validation or sizing pass, since a fitting total guarantees no individual action exceeds the limit. Only when chunking is actually required are individual action sizes computed, and those cached sizes are reused for both oversized-action validation and chunk packing (TryAddAction now takes a precomputed size instead of recalculating it). Wire compatibility, size-boundary behavior, LargePayloads capability behavior, oversized-single-action failure behavior, retry behavior, and public API are all unchanged. Adds CompleteOrchestratorTaskWithChunkingTests.cs with focused tests for: direct-send fast path, response/action size boundaries, LargePayloads on/off with an oversized action, multi-chunk wire-compatibility field preservation, and retry-then-success behavior. Fixes #773 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: dc31b150-63de-43df-9710-417064c8b07c --- .../Grpc/GrpcDurableTaskWorker.Processor.cs | 126 ++--- ...mpleteOrchestratorTaskWithChunkingTests.cs | 473 ++++++++++++++++++ 2 files changed, 539 insertions(+), 60 deletions(-) create mode 100644 test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 6d63f6af..6618be5b 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -1098,69 +1098,15 @@ async Task CompleteOrchestratorTaskWithChunkingAsync( int maxChunkBytes, CancellationToken cancellationToken) { - // Validate that no single action exceeds the maximum chunk size - static P.TaskFailureDetails? ValidateActionsSize(IEnumerable actions, int maxChunkBytes) - { - 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, - }; - } - } - - return null; - } - - P.TaskFailureDetails? validationFailure = this.worker.grpcOptions.Capabilities.Contains(P.WorkerCapability.LargePayloads) - ? null - : ValidateActionsSize(response.Actions, maxChunkBytes); - if (validationFailure != null) - { - // Complete the orchestration with a failed status and failure details - P.OrchestratorResponse failureResponse = new() - { - InstanceId = response.InstanceId, - CompletionToken = response.CompletionToken, - OrchestrationTraceContext = response.OrchestrationTraceContext, - Actions = - { - new P.OrchestratorAction - { - CompleteOrchestration = new P.CompleteOrchestrationAction - { - OrchestrationStatus = P.OrchestrationStatus.Failed, - FailureDetails = validationFailure, - }, - }, - }, - }; - - 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 + // Helper to add an action to the current chunk if it fits, using a precomputed action size + // so we never need to call CalculateSize() on the same action more than once. static bool TryAddAction( Google.Protobuf.Collections.RepeatedField dest, P.OrchestratorAction action, + int actionSize, ref int currentSize, int maxChunkBytes) { - int actionSize = action.CalculateSize(); if (currentSize + actionSize > maxChunkBytes && currentSize > 0) { return false; @@ -1171,7 +1117,10 @@ static bool TryAddAction( return true; } - // Check if the entire response fits in one chunk + // Calculate the whole response size exactly once. If it fits in a single chunk, no + // individual action can exceed that same limit (every field, including each action, + // contributes a non-negative amount to the total), so we can send it directly without + // any additional per-action validation or sizing pass. int totalSize = response.CalculateSize(); if (totalSize <= maxChunkBytes) { @@ -1183,9 +1132,66 @@ await this.ExecuteWithRetryAsync( return; } + // Response is too large to fit in a single chunk. Compute each action's serialized size + // exactly once here, and reuse the cached values below for both oversized-action + // validation and chunk packing. + List allActions = response.Actions.ToList(); + int[] actionSizes = new int[allActions.Count]; + for (int i = 0; i < allActions.Count; i++) + { + actionSizes[i] = allActions[i].CalculateSize(); + } + + if (!this.worker.grpcOptions.Capabilities.Contains(P.WorkerCapability.LargePayloads)) + { + // Validate that no single action exceeds the maximum chunk size, using the cached sizes. + for (int i = 0; i < allActions.Count; i++) + { + if (actionSizes[i] > maxChunkBytes) + { + P.OrchestratorAction action = allActions[i]; + + // 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: {actionSizes[i] / 1024.0 / 1024.0:F2}MB. " + + "Enable large-payload externalization to Azure Blob Storage to support oversized actions."; + P.TaskFailureDetails validationFailure = new() + { + ErrorType = typeof(InvalidOperationException).FullName, + ErrorMessage = errorMessage, + IsNonRetriable = true, + }; + + // Complete the orchestration with a failed status and failure details + P.OrchestratorResponse failureResponse = new() + { + InstanceId = response.InstanceId, + CompletionToken = response.CompletionToken, + OrchestrationTraceContext = response.OrchestrationTraceContext, + Actions = + { + new P.OrchestratorAction + { + CompleteOrchestration = new P.CompleteOrchestrationAction + { + OrchestrationStatus = P.OrchestrationStatus.Failed, + FailureDetails = validationFailure, + }, + }, + }, + }; + + await this.ExecuteWithRetryAsync( + async () => await this.client.CompleteOrchestratorTaskAsync(failureResponse, cancellationToken: cancellationToken), + nameof(this.client.CompleteOrchestratorTaskAsync), + cancellationToken); + return; + } + } + } + // Response is too large, split into multiple chunks int actionsCompletedSoFar = 0, chunkIndex = 0; - List allActions = response.Actions.ToList(); bool isPartial = true; bool isChunkedMode = false; @@ -1204,7 +1210,7 @@ await this.ExecuteWithRetryAsync( // Fill the chunk with actions until we reach the size limit while (actionsCompletedSoFar < allActions.Count && - TryAddAction(chunkedResponse.Actions, allActions[actionsCompletedSoFar], ref chunkPayloadSize, maxChunkBytes)) + TryAddAction(chunkedResponse.Actions, allActions[actionsCompletedSoFar], actionSizes[actionsCompletedSoFar], ref chunkPayloadSize, maxChunkBytes)) { actionsCompletedSoFar++; } diff --git a/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs b/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs new file mode 100644 index 00000000..57379280 --- /dev/null +++ b/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs @@ -0,0 +1,473 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Reflection; +using FluentAssertions; +using Grpc.Core; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.Grpc; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Moq; +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Worker.Grpc.Tests; + +/// +/// Focused unit tests for Processor.CompleteOrchestratorTaskWithChunkingAsync, covering the +/// size-boundary decisions, the capability combinations, +/// oversized-single-action failure behavior, chunk wire-compatibility, and retry behavior. These tests +/// guard against regressions in the optimization that avoids recalculating protobuf action sizes +/// (see https://github.com/microsoft/durabletask-dotnet/issues/773). +/// +public class CompleteOrchestratorTaskWithChunkingTests +{ + [Fact] + public async Task ResponseFitsExactlyAtLimit_SendsOriginalResponseDirectly_NoChunking() + { + // Arrange + P.OrchestratorResponse response = BuildResponse( + "instance-1", + BuildScheduleTaskAction(0, 16), + BuildScheduleTaskAction(1, 16)); + int maxChunkBytes = response.CalculateSize(); + + using Fixture fixture = Fixture.Create(); + + // Act + await fixture.InvokeAsync(response, maxChunkBytes); + + // Assert - the exact same response instance is sent, unmodified, in a single call. + fixture.Sent.Should().HaveCount(1); + fixture.Sent[0].Should().BeSameAs(response); +#pragma warning disable CS0612 // IsPartial/ChunkIndex are deprecated but still part of the wire contract. + fixture.Sent[0].IsPartial.Should().BeFalse(); + fixture.Sent[0].ChunkIndex.Should().BeNull(); +#pragma warning restore CS0612 + } + + [Fact] + public async Task ResponseOneByteOverLimit_ActionsStillFitIndividually_ProducesSingleNonPartialChunk() + { + // Arrange - the whole response is one byte too large, but rebuilding a chunk from the + // (already fitting) individual actions still yields just one, non-partial chunk. This + // exercises the chunking code path (not the direct-send fast path) while confirming the + // resulting chunk still preserves all wire-compatibility fields correctly. + P.OrchestratorResponse response = BuildResponse( + "instance-2", + BuildScheduleTaskAction(0, 16), + BuildScheduleTaskAction(1, 16)); + response.CustomStatus = "status"; + response.OrchestrationTraceContext = new P.OrchestrationTraceContext { SpanID = "span-1" }; + int exactSize = response.CalculateSize(); + + using Fixture fixture = Fixture.Create(); + + // Act + await fixture.InvokeAsync(response, exactSize - 1); + + // Assert + fixture.Sent.Should().HaveCount(1); + P.OrchestratorResponse sent = fixture.Sent[0]; + sent.Should().NotBeSameAs(response); // Went through the chunking construction path. + sent.InstanceId.Should().Be(response.InstanceId); + sent.CompletionToken.Should().Be(response.CompletionToken); + sent.CustomStatus.Should().Be(response.CustomStatus); + sent.OrchestrationTraceContext.SpanID.Should().Be(response.OrchestrationTraceContext.SpanID); + sent.NumEventsProcessed.Should().BeNull(); + sent.Actions.Should().HaveCount(2); +#pragma warning disable CS0612 + sent.IsPartial.Should().BeFalse(); + sent.ChunkIndex.Should().BeNull(); +#pragma warning restore CS0612 + } + + [Fact] + public async Task ActionExactlyFillsRemainingChunkSpace_IsIncludedInSameChunk() + { + // Arrange - two actions whose combined size exactly equals maxChunkBytes. The boundary + // condition in TryAddAction (currentSize + actionSize > maxChunkBytes) must treat "equal" + // as fitting, so both actions land in the same, single, non-partial chunk. + P.OrchestratorAction action0 = BuildScheduleTaskAction(0, 16); + P.OrchestratorAction action1 = BuildScheduleTaskAction(1, 32); + int size0 = action0.CalculateSize(); + int size1 = action1.CalculateSize(); + + P.OrchestratorResponse response = BuildResponse("instance-3", action0, action1); + using Fixture fixture = Fixture.Create(); + + // Act + await fixture.InvokeAsync(response, size0 + size1); + + // Assert + fixture.Sent.Should().HaveCount(1); + fixture.Sent[0].Actions.Should().HaveCount(2); + fixture.Sent[0].Actions[0].Id.Should().Be(0); + fixture.Sent[0].Actions[1].Id.Should().Be(1); +#pragma warning disable CS0612 + fixture.Sent[0].IsPartial.Should().BeFalse(); +#pragma warning restore CS0612 + } + + [Fact] + public async Task ActionExceedsRemainingChunkSpaceByOneByte_IsMovedToNextChunk() + { + // Arrange - same two actions, but maxChunkBytes is one byte less than their combined size. + // The second action must now overflow into its own, second chunk. + P.OrchestratorAction action0 = BuildScheduleTaskAction(0, 16); + P.OrchestratorAction action1 = BuildScheduleTaskAction(1, 32); + int size0 = action0.CalculateSize(); + int size1 = action1.CalculateSize(); + + P.OrchestratorResponse response = BuildResponse("instance-4", action0, action1); + using Fixture fixture = Fixture.Create(); + + // Act + await fixture.InvokeAsync(response, size0 + size1 - 1); + + // Assert + fixture.Sent.Should().HaveCount(2); + fixture.Sent[0].Actions.Should().ContainSingle(a => a.Id == 0); + fixture.Sent[1].Actions.Should().ContainSingle(a => a.Id == 1); +#pragma warning disable CS0612 + fixture.Sent[0].IsPartial.Should().BeTrue(); + fixture.Sent[0].ChunkIndex.Should().Be(0); + fixture.Sent[1].IsPartial.Should().BeFalse(); + fixture.Sent[1].ChunkIndex.Should().Be(1); +#pragma warning restore CS0612 + } + + [Fact] + public async Task NoLargePayloadsCapability_SingleOversizedAction_ReturnsFailedCompletionResponse() + { + // Arrange - one small action and one large action that alone exceeds maxChunkBytes. Without + // the LargePayloads capability, this must fail the orchestration instead of sending the + // oversized action. + P.OrchestratorAction small = BuildScheduleTaskAction(0, 16); + P.OrchestratorAction large = BuildScheduleTaskAction(1, 2048); + int largeSize = large.CalculateSize(); + int maxChunkBytes = largeSize - 1; + + P.OrchestratorResponse response = BuildResponse("instance-5", small, large); + response.OrchestrationTraceContext = new P.OrchestrationTraceContext { SpanID = "span-5" }; + using Fixture fixture = Fixture.Create(largePayloads: false); + + // Act + await fixture.InvokeAsync(response, maxChunkBytes); + + // Assert + fixture.Sent.Should().HaveCount(1); + P.OrchestratorResponse failure = fixture.Sent[0]; + failure.InstanceId.Should().Be(response.InstanceId); + failure.CompletionToken.Should().Be(response.CompletionToken); + failure.OrchestrationTraceContext.SpanID.Should().Be("span-5"); + failure.Actions.Should().HaveCount(1); + P.OrchestratorAction failureAction = failure.Actions[0]; + failureAction.CompleteOrchestration.Should().NotBeNull(); + failureAction.CompleteOrchestration.OrchestrationStatus.Should().Be(P.OrchestrationStatus.Failed); + failureAction.CompleteOrchestration.FailureDetails.IsNonRetriable.Should().BeTrue(); + failureAction.CompleteOrchestration.FailureDetails.ErrorType.Should().Be(typeof(InvalidOperationException).FullName); + string expectedMessage = $"A single orchestrator action of type ScheduleTask with id 1 " + + $"exceeds the {maxChunkBytes / 1024.0 / 1024.0:F2}MB limit: {largeSize / 1024.0 / 1024.0:F2}MB. " + + "Enable large-payload externalization to Azure Blob Storage to support oversized actions."; + failureAction.CompleteOrchestration.FailureDetails.ErrorMessage.Should().Be(expectedMessage); + } + + [Fact] + public async Task LargePayloadsCapability_SingleOversizedAction_IsSentInsteadOfFailing() + { + // Arrange - same oversized action, but with LargePayloads capability announced. The action + // must be allowed through (as its own chunk) rather than failing the orchestration. + P.OrchestratorAction small = BuildScheduleTaskAction(0, 16); + P.OrchestratorAction large = BuildScheduleTaskAction(1, 2048); + int largeSize = large.CalculateSize(); + int maxChunkBytes = largeSize - 1; + + P.OrchestratorResponse response = BuildResponse("instance-6", small, large); + using Fixture fixture = Fixture.Create(largePayloads: true); + + // Act + await fixture.InvokeAsync(response, maxChunkBytes); + + // Assert - both actions are sent (each too big to share a chunk with the other), and + // neither call carries a failed CompleteOrchestration action. + fixture.Sent.Should().HaveCountGreaterOrEqualTo(1); + fixture.Sent.SelectMany(r => r.Actions).Should().Contain(a => a.Id == 0); + fixture.Sent.SelectMany(r => r.Actions).Should().Contain(a => a.Id == 1); + fixture.Sent.SelectMany(r => r.Actions).Should().NotContain(a => a.OrchestratorActionTypeCase == P.OrchestratorAction.OrchestratorActionTypeOneofCase.CompleteOrchestration); + } + + [Fact] + public async Task MultiChunkResponse_PreservesWireCompatibilityFieldsAcrossChunks() + { + // Arrange - three actions that must span three separate chunks, verifying InstanceId, + // CompletionToken, and CustomStatus repeat every chunk; OrchestrationTraceContext is only + // set on the first chunk; NumEventsProcessed is null on the first chunk and 0 afterward; + // RequiresHistory is preserved; and chunk indices/IsPartial sequence correctly. + P.OrchestratorAction action0 = BuildScheduleTaskAction(0, 16); + P.OrchestratorAction action1 = BuildScheduleTaskAction(1, 16); + P.OrchestratorAction action2 = BuildScheduleTaskAction(2, 16); + int maxChunkBytes = Math.Max(action0.CalculateSize(), Math.Max(action1.CalculateSize(), action2.CalculateSize())); + + P.OrchestratorResponse response = BuildResponse("instance-7", action0, action1, action2); + response.CustomStatus = "custom-status"; + response.RequiresHistory = true; + response.OrchestrationTraceContext = new P.OrchestrationTraceContext { SpanID = "span-7" }; + + using Fixture fixture = Fixture.Create(); + + // Act + await fixture.InvokeAsync(response, maxChunkBytes); + + // Assert + fixture.Sent.Should().HaveCount(3); + for (int i = 0; i < fixture.Sent.Count; i++) + { + P.OrchestratorResponse chunk = fixture.Sent[i]; + chunk.InstanceId.Should().Be("instance-7"); + chunk.CompletionToken.Should().Be(response.CompletionToken); + chunk.CustomStatus.Should().Be("custom-status"); + chunk.RequiresHistory.Should().BeTrue(); +#pragma warning disable CS0612 + chunk.ChunkIndex.Should().Be(i); + chunk.IsPartial.Should().Be(i < fixture.Sent.Count - 1); +#pragma warning restore CS0612 + + if (i == 0) + { + chunk.NumEventsProcessed.Should().BeNull(); + chunk.OrchestrationTraceContext.Should().NotBeNull(); + chunk.OrchestrationTraceContext.SpanID.Should().Be("span-7"); + } + else + { + chunk.NumEventsProcessed.Should().Be(0); + chunk.OrchestrationTraceContext.Should().BeNull(); + } + } + + // All three actions were sent, in order, across the chunks, with none duplicated or dropped. + fixture.Sent.SelectMany(r => r.Actions).Select(a => a.Id).Should().Equal(0, 1, 2); + } + + [Fact] + public async Task TransientRpcError_DuringSend_RetriesAndEventuallySucceeds() + { + // Arrange - a response that fits in a single chunk (fast path), whose first send attempt + // fails with a transient gRPC error. The method must still rely on ExecuteWithRetryAsync to + // retry the same request and eventually succeed. + P.OrchestratorResponse response = BuildResponse("instance-8", BuildScheduleTaskAction(0, 16)); + int maxChunkBytes = response.CalculateSize(); + + using Fixture fixture = Fixture.Create(transientRetryBackoffBase: TimeSpan.FromMilliseconds(1)); + fixture.FailNextAttempts(1); + + // Act + await fixture.InvokeAsync(response, maxChunkBytes); + + // Assert + fixture.AttemptCount.Should().Be(2); + fixture.Sent.Should().HaveCount(1); + fixture.Sent[0].Should().BeSameAs(response); + } + + static P.OrchestratorResponse BuildResponse(string instanceId, params P.OrchestratorAction[] actions) + { + P.OrchestratorResponse response = new() + { + InstanceId = instanceId, + CompletionToken = Guid.NewGuid().ToString("N"), + }; + response.Actions.AddRange(actions); + return response; + } + + static P.OrchestratorAction BuildScheduleTaskAction(int id, int payloadBytes) + { + return new P.OrchestratorAction + { + Id = id, + ScheduleTask = new P.ScheduleTaskAction + { + Name = "Echo", + Input = new string('x', payloadBytes), + }, + }; + } + + /// + /// Test fixture that constructs a real GrpcDurableTaskWorker.Processor via reflection (it + /// is a private nested type) wired to a strictly-mocked gRPC client, and exposes a helper to + /// invoke the private CompleteOrchestratorTaskWithChunkingAsync method directly. + /// + sealed class Fixture : IDisposable + { + static readonly MethodInfo Method = FindMethod(); + + readonly object processor; + readonly object gate = new(); + int attemptsToFail; + + Fixture(object processor, Mock clientMock) + { + this.processor = processor; + this.ClientMock = clientMock; + } + + public Mock ClientMock { get; } + + public List Sent { get; } = new(); + + public int AttemptCount { get; private set; } + + public static Fixture Create( + bool largePayloads = false, + TimeSpan? transientRetryBackoffBase = null) + { + GrpcDurableTaskWorkerOptions grpcOptionsValue = new(); + if (largePayloads) + { + grpcOptionsValue.Capabilities.Add(P.WorkerCapability.LargePayloads); + } + + if (transientRetryBackoffBase.HasValue) + { + grpcOptionsValue.Internal.TransientRetryBackoffBase = transientRetryBackoffBase.Value; + } + + OptionsMonitorStub grpcOptions = new(grpcOptionsValue); + OptionsMonitorStub workerOptions = new(new DurableTaskWorkerOptions()); + Mock factoryMock = new(MockBehavior.Strict); + + GrpcDurableTaskWorker worker = new( + name: "Test", + factory: factoryMock.Object, + grpcOptions: grpcOptions, + workerOptions: workerOptions, + services: Mock.Of(), + loggerFactory: NullLoggerFactory.Instance, + orchestrationFilter: null, + exceptionPropertiesProvider: null); + + CallInvoker callInvoker = Mock.Of(); + Mock clientMock = new( + MockBehavior.Strict, new object[] { callInvoker }); + + Type processorType = typeof(GrpcDurableTaskWorker).GetNestedType("Processor", BindingFlags.NonPublic)!; + object processorInstance = Activator.CreateInstance( + processorType, + BindingFlags.Public | BindingFlags.Instance, + binder: null, + args: new object?[] { worker, clientMock.Object, null, null }, + culture: null)!; + + Fixture fixture = new(processorInstance, clientMock); + + clientMock + .Setup(c => c.CompleteOrchestratorTaskAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Returns((P.OrchestratorResponse r, Metadata h, DateTime? d, CancellationToken ct) => + fixture.HandleSend(r)); + + return fixture; + } + + /// + /// Configures the next send attempts (across all chunks) to fail + /// with a transient (Unavailable) gRPC error before subsequent attempts succeed. + /// + public void FailNextAttempts(int count) + { + lock (this.gate) + { + this.attemptsToFail = count; + } + } + + public Task InvokeAsync(P.OrchestratorResponse response, int maxChunkBytes, CancellationToken cancellationToken = default) + { + return (Task)Method.Invoke(this.processor, new object?[] { response, maxChunkBytes, cancellationToken })!; + } + + public void Dispose() + { + } + + AsyncUnaryCall HandleSend(P.OrchestratorResponse response) + { + bool shouldFail; + lock (this.gate) + { + this.AttemptCount++; + shouldFail = this.attemptsToFail > 0; + if (shouldFail) + { + this.attemptsToFail--; + } + } + + if (shouldFail) + { + return RpcExceptionAsyncUnaryCall(StatusCode.Unavailable); + } + + this.Sent.Add(response); + return CompletedAsyncUnaryCall(new P.CompleteTaskResponse()); + } + + static MethodInfo FindMethod() + { + Type processorType = typeof(GrpcDurableTaskWorker).GetNestedType("Processor", BindingFlags.NonPublic)!; + return processorType.GetMethod("CompleteOrchestratorTaskWithChunkingAsync", BindingFlags.Instance | BindingFlags.NonPublic)!; + } + + static AsyncUnaryCall CompletedAsyncUnaryCall(T response) + { + Task respTask = Task.FromResult(response); + return new AsyncUnaryCall( + respTask, + Task.FromResult(new Metadata()), + () => new Status(StatusCode.OK, string.Empty), + () => new Metadata(), + () => { }); + } + + static AsyncUnaryCall RpcExceptionAsyncUnaryCall(StatusCode statusCode, string detail = "transient error") + { + RpcException ex = new(new Status(statusCode, detail)); + Task respTask = Task.FromException(ex); + return new AsyncUnaryCall( + respTask, + Task.FromResult(new Metadata()), + () => new Status(statusCode, detail), + () => new Metadata(), + () => { }); + } + + sealed class OptionsMonitorStub : IOptionsMonitor + where T : class, new() + { + readonly T value; + + public OptionsMonitorStub(T value) => this.value = value; + + public T CurrentValue => this.value; + + public T Get(string? name) => this.value; + + public IDisposable OnChange(Action listener) => NullDisposable.Instance; + + sealed class NullDisposable : IDisposable + { + public static readonly NullDisposable Instance = new(); + + public void Dispose() + { + } + } + } + } +} From e68bbe050fb5c77a9d07356a737ee49cd958a3ce Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 16:26:04 -0700 Subject: [PATCH 2/5] Fix fail-fast regression in chunking validation ordering Restore fail-fast behavior for the no-LargePayloads case: validate each action's size in order and return immediately on the first action exceeding maxChunkBytes, without sizing later actions or computing the whole-response size first. This matches the original (pre-optimization) per-action-validation-first ordering while still reusing the sizes computed during validation for chunk packing afterward, so the issue #773 optimization is preserved for all non-failure cases (LargePayloads enabled, or response fits in one chunk, or chunking succeeds). Adds two regression tests: - Asserts the failure references the first oversized action (id=0) even when several later actions are also individually oversized, proving iteration stops at the first offender. - A calibrated timing-based test using a large shared string across 200 later actions (~1000ms to size all vs ~4ms for one) asserting the fail-fast path completes in a small fraction of that time. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: dc31b150-63de-43df-9710-417064c8b07c --- .../Grpc/GrpcDurableTaskWorker.Processor.cs | 77 +++++++++++-------- ...mpleteOrchestratorTaskWithChunkingTests.cs | 77 +++++++++++++++++++ 2 files changed, 122 insertions(+), 32 deletions(-) diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 6618be5b..29837296 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -1117,43 +1117,27 @@ static bool TryAddAction( return true; } - // Calculate the whole response size exactly once. If it fits in a single chunk, no - // individual action can exceed that same limit (every field, including each action, - // contributes a non-negative amount to the total), so we can send it directly without - // any additional per-action validation or sizing pass. - 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; - } - - // Response is too large to fit in a single chunk. Compute each action's serialized size - // exactly once here, and reuse the cached values below for both oversized-action - // validation and chunk packing. - List allActions = response.Actions.ToList(); - int[] actionSizes = new int[allActions.Count]; - for (int i = 0; i < allActions.Count; i++) + bool largePayloads = this.worker.grpcOptions.Capabilities.Contains(P.WorkerCapability.LargePayloads); + + // When the LargePayloads capability is not present, validate that no single action + // exceeds the maximum chunk size *before* doing any whole-response sizing. This must + // stay fail-fast: as soon as an oversized action is found, we fail the orchestration + // immediately without computing the size of any later action. Only when every action + // is confirmed to individually fit do we keep the sizes we already computed here, so + // they can be reused below instead of being recalculated. + int[]? actionSizes = null; + if (!largePayloads) { - actionSizes[i] = allActions[i].CalculateSize(); - } - - if (!this.worker.grpcOptions.Capabilities.Contains(P.WorkerCapability.LargePayloads)) - { - // Validate that no single action exceeds the maximum chunk size, using the cached sizes. - for (int i = 0; i < allActions.Count; i++) + actionSizes = new int[response.Actions.Count]; + for (int i = 0; i < response.Actions.Count; i++) { - if (actionSizes[i] > maxChunkBytes) + P.OrchestratorAction action = response.Actions[i]; + int actionSize = action.CalculateSize(); + if (actionSize > maxChunkBytes) { - P.OrchestratorAction action = allActions[i]; - // 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: {actionSizes[i] / 1024.0 / 1024.0:F2}MB. " + + $"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."; P.TaskFailureDetails validationFailure = new() { @@ -1187,6 +1171,35 @@ await this.ExecuteWithRetryAsync( cancellationToken); return; } + + actionSizes[i] = actionSize; + } + } + + // Calculate the whole response size exactly once. If it fits in a single chunk, send + // it directly without any further per-action work. + 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; + } + + // Response is too large to fit in a single chunk. Reuse the action sizes computed + // above during validation if we have them (LargePayloads not present); otherwise (no + // validation was needed) compute each action's serialized size exactly once here. Either + // way, the cached sizes are reused for chunk packing below instead of being recalculated. + List allActions = response.Actions.ToList(); + if (actionSizes == null) + { + actionSizes = new int[allActions.Count]; + for (int i = 0; i < allActions.Count; i++) + { + actionSizes[i] = allActions[i].CalculateSize(); } } diff --git a/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs b/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs index 57379280..ea0ca9f7 100644 --- a/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs +++ b/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs @@ -173,6 +173,83 @@ public async Task NoLargePayloadsCapability_SingleOversizedAction_ReturnsFailedC failureAction.CompleteOrchestration.FailureDetails.ErrorMessage.Should().Be(expectedMessage); } + [Fact] + public async Task NoLargePayloadsCapability_FirstActionOversized_FailsOnFirstOffender_DoesNotEvaluateLaterActions() + { + // Arrange - regression coverage for a fail-fast ordering bug: without the LargePayloads + // capability, validation must stop at the *first* oversized action instead of first sizing + // the whole response and/or every action. Action id=0 is oversized, and several later + // actions (ids 1-3) are ALSO individually oversized with distinguishable ids. If the + // algorithm regresses back to "size everything, then scan for a failure", it could still + // produce *a* failure, but this asserts it fails on id=0 specifically - proving iteration + // stopped at the very first offender rather than continuing to size/scan later actions. + P.OrchestratorAction action0 = BuildScheduleTaskAction(0, 2048); // oversized - first, must win + P.OrchestratorAction action1 = BuildScheduleTaskAction(1, 2048); // also oversized + P.OrchestratorAction action2 = BuildScheduleTaskAction(2, 2048); // also oversized + P.OrchestratorAction action3 = BuildScheduleTaskAction(3, 2048); // also oversized + int maxChunkBytes = action0.CalculateSize() - 1; + + P.OrchestratorResponse response = BuildResponse("instance-9", action0, action1, action2, action3); + using Fixture fixture = Fixture.Create(largePayloads: false); + + // Act + await fixture.InvokeAsync(response, maxChunkBytes); + + // Assert - exactly one response was sent (no partial chunks for the other oversized + // actions), and it is a failure referencing id=0. + fixture.Sent.Should().HaveCount(1); + P.OrchestratorResponse failure = fixture.Sent[0]; + failure.Actions.Should().HaveCount(1); + P.OrchestratorAction failureAction = failure.Actions[0]; + failureAction.CompleteOrchestration.Should().NotBeNull(); + failureAction.CompleteOrchestration.OrchestrationStatus.Should().Be(P.OrchestrationStatus.Failed); + failureAction.CompleteOrchestration.FailureDetails.ErrorMessage.Should().Contain("with id 0 "); + failureAction.CompleteOrchestration.FailureDetails.ErrorMessage.Should().NotContain("with id 1 "); + failureAction.CompleteOrchestration.FailureDetails.ErrorMessage.Should().NotContain("with id 2 "); + failureAction.CompleteOrchestration.FailureDetails.ErrorMessage.Should().NotContain("with id 3 "); + } + + [Fact] + public async Task NoLargePayloadsCapability_FirstActionOversized_ShortCircuitsBeforeSizingLaterActions() + { + // Arrange - a stronger, timing-based guard for the same regression as above. Action id=0 is + // oversized (tiny payload, cheap to size); actions after it each carry a large *shared* + // string reference (so memory stays low - only one big string is allocated - while + // CalculateSize() must still independently re-walk it on every call). Calibration measured + // ~1000ms to size 200 such actions vs. ~4ms to size a single one, so a fail-fast + // implementation (sizing only id=0 before returning) should complete orders of magnitude + // faster than a regressed implementation that sizes the whole response and/or every action + // first. The assertion uses a very generous threshold to avoid CI flakiness while still + // clearly separating the two behaviors. + string bigSharedInput = new('x', 40_000_000); + P.OrchestratorAction action0 = BuildScheduleTaskAction(0, 16); // small, oversized only relative to maxChunkBytes below + int maxChunkBytes = action0.CalculateSize() - 1; + + List actions = new() { action0 }; + for (int i = 1; i <= 200; i++) + { + actions.Add(new P.OrchestratorAction + { + Id = i, + ScheduleTask = new P.ScheduleTaskAction { Name = "Echo", Input = bigSharedInput }, + }); + } + + P.OrchestratorResponse response = BuildResponse("instance-10", actions.ToArray()); + using Fixture fixture = Fixture.Create(largePayloads: false); + + // Act + System.Diagnostics.Stopwatch stopwatch = System.Diagnostics.Stopwatch.StartNew(); + await fixture.InvokeAsync(response, maxChunkBytes); + stopwatch.Stop(); + + // Assert - failed fast on id=0 well before the ~1000ms it would take to size the 200 + // large later actions. + fixture.Sent.Should().HaveCount(1); + fixture.Sent[0].Actions[0].CompleteOrchestration.FailureDetails.ErrorMessage.Should().Contain("with id 0 "); + stopwatch.ElapsedMilliseconds.Should().BeLessThan(500); + } + [Fact] public async Task LargePayloadsCapability_SingleOversizedAction_IsSentInsteadOfFailing() { From c91c4a793345bf30a4f5d346ef1a9db150c46d4d Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 16:57:51 -0700 Subject: [PATCH 3/5] Replace flaky timing test with deterministic sizing-hook test The previous fail-fast regression test allocated a large shared string and asserted a wall-clock threshold, which can flake under GC/CI contention. Add a minimal test-only hook invoked immediately after each action's size is computed in the no-LargePayloads validation loop (always null in production, zero overhead beyond a null check), and rewrite the regression test to use it: it now asserts deterministically that only the first oversized action's id was ever sized, even when several later actions are also individually oversized. No wall-clock or giant-allocation assertions remain. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: dc31b150-63de-43df-9710-417064c8b07c --- .../Grpc/GrpcDurableTaskWorker.Processor.cs | 12 +++ ...mpleteOrchestratorTaskWithChunkingTests.cs | 74 +++++++++++-------- 2 files changed, 56 insertions(+), 30 deletions(-) diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 29837296..41e5a2ef 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -32,6 +32,17 @@ class Processor { static readonly Google.Protobuf.WellKnownTypes.Empty EmptyMessage = new(); + /// + /// Test-only hook invoked immediately after an individual action's size is computed during the + /// fail-fast validation pass in . Always + /// in production (adds no overhead beyond a null check); used exclusively + /// by unit tests (accessed via reflection) to deterministically verify that validation stops at + /// the first oversized action instead of computing the size of every action. + /// +#pragma warning disable CS0649 // Field is never assigned to in production code - it is set via reflection by tests. + static Action? testActionSizedHook; +#pragma warning restore CS0649 + readonly GrpcDurableTaskWorker worker; readonly TaskHubSidecarServiceClient client; readonly DurableTaskShimFactory shimFactory; @@ -1133,6 +1144,7 @@ static bool TryAddAction( { P.OrchestratorAction action = response.Actions[i]; int actionSize = action.CalculateSize(); + testActionSizedHook?.Invoke(action.Id); if (actionSize > maxChunkBytes) { // TODO: large payload doc is not available yet on aka.ms, add doc link to below error message diff --git a/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs b/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs index ea0ca9f7..a2c15b7f 100644 --- a/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs +++ b/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs @@ -210,44 +210,40 @@ public async Task NoLargePayloadsCapability_FirstActionOversized_FailsOnFirstOff } [Fact] - public async Task NoLargePayloadsCapability_FirstActionOversized_ShortCircuitsBeforeSizingLaterActions() + public async Task NoLargePayloadsCapability_FirstActionOversized_DoesNotSizeLaterActions() { - // Arrange - a stronger, timing-based guard for the same regression as above. Action id=0 is - // oversized (tiny payload, cheap to size); actions after it each carry a large *shared* - // string reference (so memory stays low - only one big string is allocated - while - // CalculateSize() must still independently re-walk it on every call). Calibration measured - // ~1000ms to size 200 such actions vs. ~4ms to size a single one, so a fail-fast - // implementation (sizing only id=0 before returning) should complete orders of magnitude - // faster than a regressed implementation that sizes the whole response and/or every action - // first. The assertion uses a very generous threshold to avoid CI flakiness while still - // clearly separating the two behaviors. - string bigSharedInput = new('x', 40_000_000); - P.OrchestratorAction action0 = BuildScheduleTaskAction(0, 16); // small, oversized only relative to maxChunkBytes below + // Arrange - deterministic (non-timing) proof of the same regression covered above: a + // test-only instrumentation hook (Processor.testActionSizedHook) records the id of every + // action whose size is actually computed during fail-fast validation. Action id=0 is + // oversized; ids 1-3 are ALSO individually oversized, so a regressed implementation that + // sizes every action up front (or continues scanning after finding one offender) would + // still record ids 1-3 here even though it happens to fail on id=0 first. Asserting the + // hook recorded *only* id=0 proves validation returned before ever sizing later actions. + P.OrchestratorAction action0 = BuildScheduleTaskAction(0, 2048); // oversized - first, must win + P.OrchestratorAction action1 = BuildScheduleTaskAction(1, 2048); // also oversized + P.OrchestratorAction action2 = BuildScheduleTaskAction(2, 2048); // also oversized + P.OrchestratorAction action3 = BuildScheduleTaskAction(3, 2048); // also oversized int maxChunkBytes = action0.CalculateSize() - 1; - List actions = new() { action0 }; - for (int i = 1; i <= 200; i++) - { - actions.Add(new P.OrchestratorAction - { - Id = i, - ScheduleTask = new P.ScheduleTaskAction { Name = "Echo", Input = bigSharedInput }, - }); - } - - P.OrchestratorResponse response = BuildResponse("instance-10", actions.ToArray()); + P.OrchestratorResponse response = BuildResponse("instance-10", action0, action1, action2, action3); using Fixture fixture = Fixture.Create(largePayloads: false); - // Act - System.Diagnostics.Stopwatch stopwatch = System.Diagnostics.Stopwatch.StartNew(); - await fixture.InvokeAsync(response, maxChunkBytes); - stopwatch.Stop(); + List sizedActionIds = new(); + Fixture.SetActionSizedHook(sizedActionIds.Add); + try + { + // Act + await fixture.InvokeAsync(response, maxChunkBytes); + } + finally + { + Fixture.SetActionSizedHook(null); + } - // Assert - failed fast on id=0 well before the ~1000ms it would take to size the 200 - // large later actions. + // Assert - only the first action's size was ever computed; ids 1-3 were never touched. + sizedActionIds.Should().Equal(0); fixture.Sent.Should().HaveCount(1); fixture.Sent[0].Actions[0].CompleteOrchestration.FailureDetails.ErrorMessage.Should().Contain("with id 0 "); - stopwatch.ElapsedMilliseconds.Should().BeLessThan(500); } [Fact] @@ -380,6 +376,7 @@ static P.OrchestratorAction BuildScheduleTaskAction(int id, int payloadBytes) sealed class Fixture : IDisposable { static readonly MethodInfo Method = FindMethod(); + static readonly FieldInfo ActionSizedHookField = FindActionSizedHookField(); readonly object processor; readonly object gate = new(); @@ -397,6 +394,17 @@ sealed class Fixture : IDisposable public int AttemptCount { get; private set; } + /// + /// Sets (or clears, when passed ) the test-only static hook that + /// Processor.CompleteOrchestratorTaskWithChunkingAsync invokes immediately after + /// computing each action's size during fail-fast validation. Always reset to + /// after use to avoid leaking state into other tests. + /// + public static void SetActionSizedHook(Action? hook) + { + ActionSizedHookField.SetValue(null, hook); + } + public static Fixture Create( bool largePayloads = false, TimeSpan? transientRetryBackoffBase = null) @@ -501,6 +509,12 @@ static MethodInfo FindMethod() return processorType.GetMethod("CompleteOrchestratorTaskWithChunkingAsync", BindingFlags.Instance | BindingFlags.NonPublic)!; } + static FieldInfo FindActionSizedHookField() + { + Type processorType = typeof(GrpcDurableTaskWorker).GetNestedType("Processor", BindingFlags.NonPublic)!; + return processorType.GetField("testActionSizedHook", BindingFlags.Static | BindingFlags.NonPublic)!; + } + static AsyncUnaryCall CompletedAsyncUnaryCall(T response) { Task respTask = Task.FromResult(response); From 63cf1035a8d14c7fae69a9a8538f8ca34cb4dd1c Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 17:07:16 -0700 Subject: [PATCH 4/5] Add whole-response sizing test hook to close fail-fast coverage gap Adds a second test-only hook (testResponseSizedHook) invoked at the response.CalculateSize() call site, complementing the existing per-action sizing hook. The per-action hook alone cannot detect a regression that reintroduces whole-response sizing before fail-fast validation, since protobuf's recursive sizing bypasses it. The fail-fast test now asserts the whole-response hook is never invoked, and a new assertion on the happy-path test proves the hook fires exactly once, guarding against false-positive coverage from broken hook wiring. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: dc31b150-63de-43df-9710-417064c8b07c --- .../Grpc/GrpcDurableTaskWorker.Processor.cs | 13 ++++ ...mpleteOrchestratorTaskWithChunkingTests.cs | 63 ++++++++++++++++--- 2 files changed, 66 insertions(+), 10 deletions(-) diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 41e5a2ef..73e2b66f 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -41,6 +41,18 @@ class Processor /// #pragma warning disable CS0649 // Field is never assigned to in production code - it is set via reflection by tests. static Action? testActionSizedHook; + + /// + /// Test-only hook invoked immediately after the *whole response* is sized (via + /// response.CalculateSize()) in . + /// Always in production (adds no overhead beyond a null check). Protobuf's + /// whole-response sizing recursively sizes every action internally without going through + /// , so this separate hook exists to let unit tests prove that + /// whole-response sizing is never reached when fail-fast validation rejects an oversized action - + /// a regression that reintroduces whole-response sizing *before* validation would otherwise go + /// undetected by alone. + /// + static Action? testResponseSizedHook; #pragma warning restore CS0649 readonly GrpcDurableTaskWorker worker; @@ -1191,6 +1203,7 @@ await this.ExecuteWithRetryAsync( // Calculate the whole response size exactly once. If it fits in a single chunk, send // it directly without any further per-action work. int totalSize = response.CalculateSize(); + testResponseSizedHook?.Invoke(); if (totalSize <= maxChunkBytes) { // Response fits in one chunk, send it directly (isPartial defaults to false) diff --git a/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs b/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs index a2c15b7f..2150bb32 100644 --- a/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs +++ b/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs @@ -34,10 +34,23 @@ public async Task ResponseFitsExactlyAtLimit_SendsOriginalResponseDirectly_NoChu using Fixture fixture = Fixture.Create(); - // Act - await fixture.InvokeAsync(response, maxChunkBytes); + // Sanity-check the whole-response-sizing hook itself: it must fire exactly once for a + // normal (non-fail-fast) request, so that asserting "never invoked" in the fail-fast test + // below is meaningful rather than a false positive caused by broken hook wiring. + int responseSizedCount = 0; + Fixture.SetResponseSizedHook(() => responseSizedCount++); + try + { + // Act + await fixture.InvokeAsync(response, maxChunkBytes); + } + finally + { + Fixture.SetResponseSizedHook(null); + } // Assert - the exact same response instance is sent, unmodified, in a single call. + responseSizedCount.Should().Be(1); fixture.Sent.Should().HaveCount(1); fixture.Sent[0].Should().BeSameAs(response); #pragma warning disable CS0612 // IsPartial/ChunkIndex are deprecated but still part of the wire contract. @@ -212,13 +225,20 @@ public async Task NoLargePayloadsCapability_FirstActionOversized_FailsOnFirstOff [Fact] public async Task NoLargePayloadsCapability_FirstActionOversized_DoesNotSizeLaterActions() { - // Arrange - deterministic (non-timing) proof of the same regression covered above: a - // test-only instrumentation hook (Processor.testActionSizedHook) records the id of every - // action whose size is actually computed during fail-fast validation. Action id=0 is - // oversized; ids 1-3 are ALSO individually oversized, so a regressed implementation that - // sizes every action up front (or continues scanning after finding one offender) would - // still record ids 1-3 here even though it happens to fail on id=0 first. Asserting the - // hook recorded *only* id=0 proves validation returned before ever sizing later actions. + // Arrange - deterministic (non-timing) proof of the same regression covered above, using two + // complementary test-only instrumentation hooks: + // 1. Processor.testActionSizedHook records the id of every action whose size is actually + // computed during fail-fast validation. Action id=0 is oversized; ids 1-3 are ALSO + // individually oversized, so a regressed implementation that sizes every action up front + // (or continues scanning after finding one offender) would still record ids 1-3 here even + // though it happens to fail on id=0 first. Asserting the hook recorded *only* id=0 proves + // validation returned before ever sizing later actions via the per-action path. + // 2. Processor.testResponseSizedHook fires when the *whole response* is sized via + // response.CalculateSize(). Protobuf's whole-response sizing recursively sizes every + // action internally, bypassing hook (1) entirely - so a regression that reintroduces + // whole-response sizing *before* validation (instead of only after validation succeeds) + // would go undetected by hook (1) alone. Asserting hook (2) is never invoked proves + // whole-response sizing is not reached when validation fails fast. P.OrchestratorAction action0 = BuildScheduleTaskAction(0, 2048); // oversized - first, must win P.OrchestratorAction action1 = BuildScheduleTaskAction(1, 2048); // also oversized P.OrchestratorAction action2 = BuildScheduleTaskAction(2, 2048); // also oversized @@ -229,7 +249,9 @@ public async Task NoLargePayloadsCapability_FirstActionOversized_DoesNotSizeLate using Fixture fixture = Fixture.Create(largePayloads: false); List sizedActionIds = new(); + int responseSizedCount = 0; Fixture.SetActionSizedHook(sizedActionIds.Add); + Fixture.SetResponseSizedHook(() => responseSizedCount++); try { // Act @@ -238,10 +260,13 @@ public async Task NoLargePayloadsCapability_FirstActionOversized_DoesNotSizeLate finally { Fixture.SetActionSizedHook(null); + Fixture.SetResponseSizedHook(null); } - // Assert - only the first action's size was ever computed; ids 1-3 were never touched. + // Assert - only the first action's size was ever computed; ids 1-3 were never touched, and + // whole-response sizing was never reached (the fail-fast return happens before it). sizedActionIds.Should().Equal(0); + responseSizedCount.Should().Be(0); fixture.Sent.Should().HaveCount(1); fixture.Sent[0].Actions[0].CompleteOrchestration.FailureDetails.ErrorMessage.Should().Contain("with id 0 "); } @@ -377,6 +402,7 @@ sealed class Fixture : IDisposable { static readonly MethodInfo Method = FindMethod(); static readonly FieldInfo ActionSizedHookField = FindActionSizedHookField(); + static readonly FieldInfo ResponseSizedHookField = FindResponseSizedHookField(); readonly object processor; readonly object gate = new(); @@ -405,6 +431,17 @@ public static void SetActionSizedHook(Action? hook) ActionSizedHookField.SetValue(null, hook); } + /// + /// Sets (or clears, when passed ) the test-only static hook that + /// Processor.CompleteOrchestratorTaskWithChunkingAsync invokes immediately after + /// sizing the *whole response* via response.CalculateSize(). Always reset to + /// after use to avoid leaking state into other tests. + /// + public static void SetResponseSizedHook(Action? hook) + { + ResponseSizedHookField.SetValue(null, hook); + } + public static Fixture Create( bool largePayloads = false, TimeSpan? transientRetryBackoffBase = null) @@ -515,6 +552,12 @@ static FieldInfo FindActionSizedHookField() return processorType.GetField("testActionSizedHook", BindingFlags.Static | BindingFlags.NonPublic)!; } + static FieldInfo FindResponseSizedHookField() + { + Type processorType = typeof(GrpcDurableTaskWorker).GetNestedType("Processor", BindingFlags.NonPublic)!; + return processorType.GetField("testResponseSizedHook", BindingFlags.Static | BindingFlags.NonPublic)!; + } + static AsyncUnaryCall CompletedAsyncUnaryCall(T response) { Task respTask = Task.FromResult(response); From 9609b20d20baf3fd7b73250aab271cc66db4fe58 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 27 Jul 2026 11:36:52 -0700 Subject: [PATCH 5/5] Harden chunking performance tests Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 01dc5b7c-c742-45b7-ae8c-5cda9bcfbd2c --- .../Grpc/GrpcDurableTaskWorker.Processor.cs | 15 +++++++-------- .../CompleteOrchestratorTaskWithChunkingTests.cs | 9 +++++++++ 2 files changed, 16 insertions(+), 8 deletions(-) diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 73e2b66f..705468eb 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -1122,7 +1122,7 @@ async Task CompleteOrchestratorTaskWithChunkingAsync( CancellationToken cancellationToken) { // Helper to add an action to the current chunk if it fits, using a precomputed action size - // so we never need to call CalculateSize() on the same action more than once. + // so validation and chunk packing do not need to recalculate its serialized size. static bool TryAddAction( Google.Protobuf.Collections.RepeatedField dest, P.OrchestratorAction action, @@ -1218,13 +1218,12 @@ await this.ExecuteWithRetryAsync( // above during validation if we have them (LargePayloads not present); otherwise (no // validation was needed) compute each action's serialized size exactly once here. Either // way, the cached sizes are reused for chunk packing below instead of being recalculated. - List allActions = response.Actions.ToList(); if (actionSizes == null) { - actionSizes = new int[allActions.Count]; - for (int i = 0; i < allActions.Count; i++) + actionSizes = new int[response.Actions.Count]; + for (int i = 0; i < response.Actions.Count; i++) { - actionSizes[i] = allActions[i].CalculateSize(); + actionSizes[i] = response.Actions[i].CalculateSize(); } } @@ -1247,15 +1246,15 @@ await this.ExecuteWithRetryAsync( int chunkPayloadSize = 0; // Fill the chunk with actions until we reach the size limit - while (actionsCompletedSoFar < allActions.Count && - TryAddAction(chunkedResponse.Actions, allActions[actionsCompletedSoFar], actionSizes[actionsCompletedSoFar], ref chunkPayloadSize, maxChunkBytes)) + while (actionsCompletedSoFar < response.Actions.Count && + TryAddAction(chunkedResponse.Actions, response.Actions[actionsCompletedSoFar], actionSizes[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; + isPartial = actionsCompletedSoFar < response.Actions.Count; chunkedResponse.IsPartial = isPartial; // Only activate chunked mode when we actually need multiple chunks. diff --git a/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs b/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs index 2150bb32..3c451774 100644 --- a/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs +++ b/test/Worker/Grpc.Tests/CompleteOrchestratorTaskWithChunkingTests.cs @@ -13,6 +13,14 @@ namespace Microsoft.DurableTask.Worker.Grpc.Tests; +/// +/// Serializes tests that use the Processor's static sizing-observability hooks. +/// +[CollectionDefinition(nameof(CompleteOrchestratorTaskWithChunkingCollection), DisableParallelization = true)] +public sealed class CompleteOrchestratorTaskWithChunkingCollection +{ +} + /// /// Focused unit tests for Processor.CompleteOrchestratorTaskWithChunkingAsync, covering the /// size-boundary decisions, the capability combinations, @@ -20,6 +28,7 @@ namespace Microsoft.DurableTask.Worker.Grpc.Tests; /// guard against regressions in the optimization that avoids recalculating protobuf action sizes /// (see https://github.com/microsoft/durabletask-dotnet/issues/773). /// +[Collection(nameof(CompleteOrchestratorTaskWithChunkingCollection))] public class CompleteOrchestratorTaskWithChunkingTests { [Fact]