diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 6d63f6af..1c856505 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,41 @@ 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 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 + // 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 = pastEvents.Concat(chunk.Events.Select(converter)); + 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..816dfb87 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,189 @@ 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(); + } + + [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(); + + 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 + { + EventId = eventId, + Timestamp = Timestamp.FromDateTime(DateTime.UtcNow), + GenericEvent = new P.GenericEvent { Data = data }, + }; + } + [Fact] public void Constructor_StrictWorkerVersioningWithoutRegistryContents_DoesNotThrow() { @@ -819,6 +1007,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 +1254,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); + } + } }