From 0f43f4d7cdab2e7c3e73c6b36fbd7b4942e82c92 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 15:06:37 -0700 Subject: [PATCH 1/4] Perf: materialize streamed orchestration history incrementally (#772) Replace the lazy Concat/Select chain in BuildRuntimeStateAsync with direct List accumulation. Previously, each streamed history chunk extended a lazily-evaluated IEnumerable via Concat, which is re-walked and re-converted from scratch every time it is enumerated or chained onto again, making history reconstruction quadratic in the number of chunks for long-running orchestrations. Now each chunk's events are converted and appended directly into a single accumulating list (with a capacity hint sized per chunk), and new events are added to the runtime state via a direct foreach instead of a lazy Select. Event conversion, ordering, new-vs-past runtime state classification, cancellation propagation, and the public API surface are all unchanged. Added regression tests covering multi-chunk streamed history materialization (ordering + new/past classification) and cancellation propagation mid-stream. Fixes #772 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 1fdfdae0-4987-4043-b522-7229e63810d2 --- .../Grpc/GrpcDurableTaskWorker.Processor.cs | 27 ++- .../Grpc.Tests/GrpcDurableTaskWorkerTests.cs | 183 ++++++++++++++++++ 2 files changed, 202 insertions(+), 8 deletions(-) diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 6d63f6af..96df0371 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -256,7 +256,7 @@ async ValueTask BuildRuntimeStateAsync( ? ProtoUtils.ConvertHistoryEvent : entityConversionState.ConvertFromProto; - IEnumerable pastEvents = []; + List pastEvents; if (orchestratorRequest.RequiresHistoryStreaming) { // Stream the remaining events from the remote service @@ -270,25 +270,36 @@ async ValueTask BuildRuntimeStateAsync( using AsyncServerStreamingCall streamResponse = this.client.StreamInstanceHistory(streamRequest, cancellationToken: cancellation); + // Materialize each chunk's events directly into the accumulating list as they arrive, + // rather than lazily chaining Concat/Select over the chunks. The lazy chain would be + // re-enumerated (and re-converted) from scratch every time it is chained onto, making + // history reconstruction quadratic in the number of chunks for long-running orchestrations. + pastEvents = new List(); await foreach (P.HistoryChunk chunk in streamResponse.ResponseStream.ReadAllAsync(cancellation)) { - pastEvents = pastEvents.Concat(chunk.Events.Select(converter)); + pastEvents.Capacity = Math.Max(pastEvents.Capacity, pastEvents.Count + chunk.Events.Count); + foreach (P.HistoryEvent protoEvent in chunk.Events) + { + pastEvents.Add(converter(protoEvent)); + } } } else { // The history was already provided in the work item request - pastEvents = orchestratorRequest.PastEvents.Select(converter); + pastEvents = new List(orchestratorRequest.PastEvents.Count); + foreach (P.HistoryEvent protoEvent in orchestratorRequest.PastEvents) + { + pastEvents.Add(converter(protoEvent)); + } } - IEnumerable newEvents = orchestratorRequest.NewEvents.Select(converter); - // Reconstruct the orchestration state in a way that correctly distinguishes new events from past events - var runtimeState = new OrchestrationRuntimeState(pastEvents.ToList()); - foreach (HistoryEvent e in newEvents) + var runtimeState = new OrchestrationRuntimeState(pastEvents); + foreach (P.HistoryEvent protoEvent in orchestratorRequest.NewEvents) { // AddEvent() puts events into the NewEvents list. - runtimeState.AddEvent(e); + runtimeState.AddEvent(converter(protoEvent)); } if (runtimeState.ExecutionStartedEvent == null) diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs index bc9faab6..648711a8 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs @@ -4,6 +4,8 @@ using System.Collections.Concurrent; using System.IO; using System.Reflection; +using DurableTask.Core; +using DurableTask.Core.History; using Google.Protobuf.WellKnownTypes; using Grpc.Core; using Microsoft.DurableTask; @@ -34,6 +36,9 @@ public class GrpcDurableTaskWorkerTests .GetMethod("DispatchWorkItem", BindingFlags.Instance | BindingFlags.NonPublic)!; static readonly MethodInfo TryRecreateChannelAsyncMethod = typeof(GrpcDurableTaskWorker) .GetMethod("TryRecreateChannelAsync", BindingFlags.Instance | BindingFlags.NonPublic)!; + static readonly MethodInfo BuildRuntimeStateAsyncMethod = typeof(GrpcDurableTaskWorker) + .GetNestedType("Processor", BindingFlags.NonPublic)! + .GetMethod("BuildRuntimeStateAsync", BindingFlags.Instance | BindingFlags.NonPublic)!; [Fact] public async Task ExecuteAsync_ConnectFailureThreshold_RecreatesConfiguredChannel() @@ -598,6 +603,105 @@ public async Task ConnectAsync_VeryLargeHelloDeadline_UsesUtcMaxValueDeadline() deadline.Should().Be(DateTime.SpecifyKind(DateTime.MaxValue, DateTimeKind.Utc)); } + [Fact] + public async Task BuildRuntimeStateAsync_MultiChunkHistoryStream_MaterializesEventsInOrderAndClassifiesNewVsPast() + { + // Arrange — three chunks arriving over the StreamInstanceHistory call. Chunk boundaries are + // arbitrary from the caller's perspective; the resulting runtime state must read as if all events + // had arrived in a single chunk, in the same relative order, with past/new events kept distinct. + P.HistoryEvent executionStarted = new() + { + EventId = -1, + Timestamp = Timestamp.FromDateTime(DateTime.UtcNow), + ExecutionStarted = new P.ExecutionStartedEvent + { + Name = "TestOrchestration", + Version = "1.0", + OrchestrationInstance = new P.OrchestrationInstance + { + InstanceId = "instance-1", + ExecutionId = "execution-1", + }, + }, + }; + + P.HistoryChunk[] chunks = + { + new() { Events = { executionStarted } }, + new() { Events = { CreateGenericEvent(1, "chunk-2-event-1"), CreateGenericEvent(2, "chunk-2-event-2") } }, + new() { Events = { CreateGenericEvent(3, "chunk-3-event-1") } }, + }; + + SequenceAsyncStreamReader reader = new(chunks); + HistoryStreamCallInvoker callInvoker = new(reader); + P.TaskHubSidecarService.TaskHubSidecarServiceClient client = new(callInvoker); + GrpcDurableTaskWorker worker = CreateWorker(new GrpcDurableTaskWorkerOptions()); + object processor = CreateProcessor(worker, client); + + P.OrchestratorRequest request = new() + { + InstanceId = "instance-1", + ExecutionId = "execution-1", + RequiresHistoryStreaming = true, + }; + request.NewEvents.Add(CreateGenericEvent(4, "new-event")); + + // Act + OrchestrationRuntimeState runtimeState = await InvokeBuildRuntimeStateAsync( + processor, request, entityConversionState: null, CancellationToken.None); + + // Assert + runtimeState.PastEvents.Should().HaveCount(4); + runtimeState.PastEvents[0].Should().BeOfType(); + ((GenericEvent)runtimeState.PastEvents[1]).Data.Should().Be("chunk-2-event-1"); + ((GenericEvent)runtimeState.PastEvents[2]).Data.Should().Be("chunk-2-event-2"); + ((GenericEvent)runtimeState.PastEvents[3]).Data.Should().Be("chunk-3-event-1"); + + runtimeState.NewEvents.Should().HaveCount(1); + ((GenericEvent)runtimeState.NewEvents[0]).Data.Should().Be("new-event"); + + runtimeState.ExecutionStartedEvent.Should().NotBeNull(); + runtimeState.ExecutionStartedEvent!.Name.Should().Be("TestOrchestration"); + } + + [Fact] + public async Task BuildRuntimeStateAsync_HistoryStreamCanceledMidStream_PropagatesCancellation() + { + // Arrange — the second chunk read observes a canceled token. Cancellation must propagate out of + // BuildRuntimeStateAsync rather than being swallowed while accumulating history chunks. + using CancellationTokenSource cts = new(); + P.HistoryChunk firstChunk = new() { Events = { CreateGenericEvent(1, "chunk-1-event-1") } }; + CancelingAsyncStreamReader reader = new(cts, firstChunk); + HistoryStreamCallInvoker callInvoker = new(reader); + P.TaskHubSidecarService.TaskHubSidecarServiceClient client = new(callInvoker); + GrpcDurableTaskWorker worker = CreateWorker(new GrpcDurableTaskWorkerOptions()); + object processor = CreateProcessor(worker, client); + + P.OrchestratorRequest request = new() + { + InstanceId = "instance-1", + ExecutionId = "execution-1", + RequiresHistoryStreaming = true, + }; + + // Act + Func act = () => InvokeBuildRuntimeStateAsync( + processor, request, entityConversionState: null, cts.Token); + + // Assert + await act.Should().ThrowAsync(); + } + + static P.HistoryEvent CreateGenericEvent(int eventId, string data) + { + return new P.HistoryEvent + { + EventId = eventId, + Timestamp = Timestamp.FromDateTime(DateTime.UtcNow), + GenericEvent = new P.GenericEvent { Data = data }, + }; + } + [Fact] public void Constructor_StrictWorkerVersioningWithoutRegistryContents_DoesNotThrow() { @@ -819,6 +923,17 @@ static async Task InvokeTryRecreateChannelAsync( return task.GetType().GetProperty("Result")!.GetValue(task)!; } + static async Task InvokeBuildRuntimeStateAsync( + object processor, + P.OrchestratorRequest orchestratorRequest, + object? entityConversionState, + CancellationToken cancellationToken) + { + object result = BuildRuntimeStateAsyncMethod.Invoke( + processor, new object?[] { orchestratorRequest, entityConversionState, cancellationToken })!; + return await (ValueTask)result; + } + static T GetResultProperty(object result, string propertyName) { return (T)result.GetType().GetProperty(propertyName)!.GetValue(result)!; @@ -1055,4 +1170,72 @@ public override AsyncDuplexStreamingCall AsyncDuplexStreami throw new NotSupportedException(); } } + + sealed class HistoryStreamCallInvoker : CallInvoker + { + readonly IAsyncStreamReader reader; + + public HistoryStreamCallInvoker(IAsyncStreamReader reader) + { + this.reader = reader; + } + + public override TResponse BlockingUnaryCall(Method method, string? host, CallOptions options, TRequest request) + { + throw new NotSupportedException(); + } + + public override AsyncUnaryCall AsyncUnaryCall(Method method, string? host, CallOptions options, TRequest request) + { + throw new NotSupportedException($"Unexpected unary method {method.FullName}."); + } + + public override AsyncServerStreamingCall AsyncServerStreamingCall(Method method, string? host, CallOptions options, TRequest request) + { + if (method.FullName == "/TaskHubSidecarService/StreamInstanceHistory") + { + return (AsyncServerStreamingCall)(object)CreateServerStreamingCall(this.reader); + } + + throw new NotSupportedException($"Unexpected server-streaming method {method.FullName}."); + } + + public override AsyncClientStreamingCall AsyncClientStreamingCall(Method method, string? host, CallOptions options) + { + throw new NotSupportedException(); + } + + public override AsyncDuplexStreamingCall AsyncDuplexStreamingCall(Method method, string? host, CallOptions options) + { + throw new NotSupportedException(); + } + } + + sealed class CancelingAsyncStreamReader : IAsyncStreamReader + { + readonly CancellationTokenSource cancellationSource; + readonly Queue items; + + public CancelingAsyncStreamReader(CancellationTokenSource cancellationSource, params T[] items) + { + this.cancellationSource = cancellationSource; + this.items = new Queue(items); + } + + public T Current { get; private set; } = default!; + + public Task MoveNext(CancellationToken cancellationToken) + { + if (this.items.Count > 0) + { + this.Current = this.items.Dequeue(); + return Task.FromResult(true); + } + + // Simulate the underlying stream observing cancellation once the caller cancels mid-stream. + this.cancellationSource.Cancel(); + cancellationToken.ThrowIfCancellationRequested(); + return Task.FromResult(false); + } + } } From 9acfd585e9b4084b897e35d6980a3fdaec6f2c9e Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 16:18:21 -0700 Subject: [PATCH 2/4] Fix quadratic list growth from per-chunk Capacity forcing Address review feedback on PR #783 (issue #772): setting List.Capacity to the exact running total on every streamed chunk reallocates and copies the whole list whenever capacity needs to grow, since the Capacity setter allocates exactly the requested size rather than growing geometrically. With many single-event chunks this made history materialization quadratic again - the same class of bug the original fix addressed, reintroduced via a different mechanism. Remove the explicit Capacity assignment and rely on List.Add's own amortized (doubling) growth strategy, which gives O(n) total allocation across n appends regardless of how many chunks they arrive in. Added a regression test with 5,000 single-event chunks that asserts on bytes allocated (via GC.GetAllocatedBytesForCurrentThread) rather than wall-clock time, to avoid timing-based flakiness while still reliably distinguishing O(n) amortized growth from O(n^2) per-chunk reallocation. Verified the test fails against the prior buggy code (~100 MB allocated for n=5,000, chunk copied) and passes against the fix (well under the 50 MB threshold). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 1fdfdae0-4987-4043-b522-7229e63810d2 --- .../Grpc/GrpcDurableTaskWorker.Processor.cs | 7 +- .../Grpc.Tests/GrpcDurableTaskWorkerTests.cs | 87 +++++++++++++++++++ 2 files changed, 93 insertions(+), 1 deletion(-) diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 96df0371..bd4a89f1 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -274,10 +274,15 @@ async ValueTask BuildRuntimeStateAsync( // rather than lazily chaining Concat/Select over the chunks. The lazy chain would be // re-enumerated (and re-converted) from scratch every time it is chained onto, making // history reconstruction quadratic in the number of chunks for long-running orchestrations. + // + // Note: intentionally do NOT set List.Capacity to an exact per-chunk running total here. + // The Capacity setter reallocates to precisely the requested size, so with many small + // chunks (e.g. one event per chunk) that would reallocate and copy on every chunk, which is + // itself quadratic. List.Add's built-in geometric (doubling) growth already gives + // amortized O(1) appends, so we let it manage capacity on its own. pastEvents = new List(); await foreach (P.HistoryChunk chunk in streamResponse.ResponseStream.ReadAllAsync(cancellation)) { - pastEvents.Capacity = Math.Max(pastEvents.Capacity, pastEvents.Count + chunk.Events.Count); foreach (P.HistoryEvent protoEvent in chunk.Events) { pastEvents.Add(converter(protoEvent)); diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs index 648711a8..9693f1e1 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs @@ -692,6 +692,93 @@ public async Task BuildRuntimeStateAsync_HistoryStreamCanceledMidStream_Propagat await act.Should().ThrowAsync(); } + [Fact] + public async Task BuildRuntimeStateAsync_ManySingleEventChunks_DoesNotExhibitQuadraticListGrowth() + { + // Arrange — regression test for a prior bug where the accumulating list's Capacity was forced to + // the exact running total on every chunk. That defeats List's own amortized (geometric) + // doubling growth strategy: with many single-event chunks, it reallocates and copies the entire + // list on *every* chunk, making history materialization quadratic again (the same class of bug + // issue #772 originally fixed, just reintroduced via a different mechanism). + // + // Wall-clock timing assertions are flaky across machines/CI load, so instead this asserts on + // bytes allocated, which cleanly separates the two regimes for a large enough event count: + // amortized doubling growth allocates O(n) total across all chunks, while forcing Capacity to the + // exact count per single-event chunk allocates O(n^2) (a sum-of-1..n series of full-list copies). + const int eventCount = 5_000; + + P.HistoryEvent executionStarted = new() + { + EventId = -1, + Timestamp = Timestamp.FromDateTime(DateTime.UtcNow), + ExecutionStarted = new P.ExecutionStartedEvent + { + Name = "TestOrchestration", + Version = "1.0", + OrchestrationInstance = new P.OrchestrationInstance + { + InstanceId = "instance-1", + ExecutionId = "execution-1", + }, + }, + }; + + List chunks = new(eventCount + 1) { new() { Events = { executionStarted } } }; + for (int i = 0; i < eventCount; i++) + { + // One event per chunk is the worst case for any strategy that resizes per chunk. + chunks.Add(new P.HistoryChunk { Events = { CreateGenericEvent(i, $"event-{i}") } }); + } + + GrpcDurableTaskWorker worker = CreateWorker(new GrpcDurableTaskWorkerOptions()); + + async Task RunAsync() + { + SequenceAsyncStreamReader reader = new(chunks.ToArray()); + HistoryStreamCallInvoker callInvoker = new(reader); + P.TaskHubSidecarService.TaskHubSidecarServiceClient client = new(callInvoker); + object processor = CreateProcessor(worker, client); + + P.OrchestratorRequest request = new() + { + InstanceId = "instance-1", + ExecutionId = "execution-1", + RequiresHistoryStreaming = true, + }; + + return await InvokeBuildRuntimeStateAsync( + processor, request, entityConversionState: null, CancellationToken.None); + } + + // Warm up once (JIT/type initialization) before measuring allocations for the real run. + await RunAsync(); + + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + long allocatedBefore = GC.GetAllocatedBytesForCurrentThread(); + + // Act + OrchestrationRuntimeState runtimeState = await RunAsync(); + + long allocatedBytes = GC.GetAllocatedBytesForCurrentThread() - allocatedBefore; + + // Assert — functional correctness: all events materialized, in order. + runtimeState.PastEvents.Should().HaveCount(eventCount + 1); + runtimeState.PastEvents[0].Should().BeOfType(); + ((GenericEvent)runtimeState.PastEvents[eventCount]).Data.Should().Be($"event-{eventCount - 1}"); + + // Assert — performance: O(n) amortized list growth allocates on the order of a few hundred + // kilobytes to a few megabytes for this input size (list growth plus per-event object overhead). + // The prior per-chunk-exact-Capacity bug instead reallocates and copies the whole list on every + // single-event chunk, which allocates on the order of n^2 * pointer-size bytes (~100+ MB for + // n = 5,000) — orders of magnitude more. The threshold below is generous enough to avoid flakiness + // from ordinary object/collection overhead while remaining far below the quadratic regime. + allocatedBytes.Should().BeLessThan( + 50 * 1024 * 1024, + "history materialization should not exhibit quadratic (O(n^2)) memory growth across many streamed chunks"); + } + static P.HistoryEvent CreateGenericEvent(int eventId, string data) { return new P.HistoryEvent From 64e21904eaf4e2cff75a383f4d4dc9de4e079d48 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 27 Jul 2026 11:36:52 -0700 Subject: [PATCH 3/4] Remove forced GC from history allocation test Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 01dc5b7c-c742-45b7-ae8c-5cda9bcfbd2c --- test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs | 3 --- 1 file changed, 3 deletions(-) diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs index 9693f1e1..816dfb87 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs @@ -753,9 +753,6 @@ async Task RunAsync() // Warm up once (JIT/type initialization) before measuring allocations for the real run. await RunAsync(); - GC.Collect(); - GC.WaitForPendingFinalizers(); - GC.Collect(); long allocatedBefore = GC.GetAllocatedBytesForCurrentThread(); // Act From cfe3060c6d21b6b9ffe52724370607755cda1927 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 27 Jul 2026 16:31:44 -0700 Subject: [PATCH 4/4] Correct streamed history performance rationale Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 01dc5b7c-c742-45b7-ae8c-5cda9bcfbd2c --- src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index bd4a89f1..1c856505 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -271,9 +271,9 @@ async ValueTask BuildRuntimeStateAsync( this.client.StreamInstanceHistory(streamRequest, cancellationToken: cancellation); // Materialize each chunk's events directly into the accumulating list as they arrive, - // rather than lazily chaining Concat/Select over the chunks. The lazy chain would be - // re-enumerated (and re-converted) from scratch every time it is chained onto, making - // history reconstruction quadratic in the number of chunks for long-running orchestrations. + // rather than building a deeply nested Concat/Select iterator chain. Enumerating that + // chain dispatches through the preceding Concat nodes for each event, so with many small + // chunks the iterator work grows quadratically in the number of chunks. // // Note: intentionally do NOT set List.Capacity to an exact per-chunk running total here. // The Capacity setter reallocates to precisely the requested size, so with many small