Perf: materialize streamed orchestration history incrementally

[berndverst]

Summary

Replaces the lazy Concat/Select history accumulation in BuildRuntimeStateAsync (src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs) with direct List<HistoryEvent> accumulation.

Previously, each streamed history chunk extended a lazily-evaluated IEnumerable<HistoryEvent> via Concat. Because Concat chains are re-walked (and re-converted) from scratch every time they are enumerated or chained onto again, reconstructing history for a streamed orchestration was effectively quadratic in the number of chunks — a real problem for long-running orchestrations with large or heavily-chunked history.

Change

• Both the streaming (RequiresHistoryStreaming) and non-streaming (PastEvents) branches now build a single List<HistoryEvent> directly, appending converted events as they arrive (with a capacity hint sized per chunk in the streaming path) instead of chaining lazy Concat/Select operations.
• NewEvents are now added to the runtime state via a direct foreach + runtimeState.AddEvent(...) instead of a lazy Select projection.
• Event conversion (converter/ProtoUtils.ConvertHistoryEvent / entity conversion), event ordering, new-vs-past runtime state classification, cancellation propagation, and the public API surface are all unchanged.

Tests

Added two focused regression tests to test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs, following existing test conventions in the file:

• BuildRuntimeStateAsync_MultiChunkHistoryStream_MaterializesEventsInOrderAndClassifiesNewVsPast — streams history across three chunks and verifies events are materialized in order into PastEvents, NewEvents classification is preserved, and ExecutionStartedEvent is populated correctly.
• BuildRuntimeStateAsync_HistoryStreamCanceledMidStream_PropagatesCancellation — verifies cancellation observed mid-stream propagates as OperationCanceledException rather than being swallowed.

Targeted run:

dotnet test test\Worker\Grpc.Tests\Worker.Grpc.Tests.csproj --filter "FullyQualifiedName~BuildRuntimeStateAsync"

→ 2/2 passed.

Full project suite:

dotnet test test\Worker\Grpc.Tests\Worker.Grpc.Tests.csproj

→ 139/139 passed, 0 failures.

Fixes #772

[Copilot]

Pull request overview

This PR improves worker-side orchestration replay performance by replacing lazy IEnumerable accumulation (Concat/Select) in BuildRuntimeStateAsync with eager, incremental materialization into a single List<HistoryEvent>, preventing quadratic behavior when history is streamed in many chunks.

Changes:

• Materialize streamed and non-streamed PastEvents into a single List<HistoryEvent> as events arrive.
• Add NewEvents to OrchestrationRuntimeState via direct iteration and conversion (no lazy projections).
• Add regression tests covering multi-chunk streaming order/classification and cancellation propagation.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs	Reworks history accumulation to eagerly append converted events instead of chaining lazy Concat/Select.
test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs	Adds targeted tests for multi-chunk history materialization correctness and cancellation behavior.

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (1)

src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs:287

• In the streaming path, pastEvents starts at the default capacity and grows via repeated Add calls. For large streamed chunks, this can still trigger multiple intermediate reallocations/copies within a single chunk. Since chunk.Events.Count is available, consider calling List<T>.EnsureCapacity(pastEvents.Count + chunk.Events.Count) per chunk to reduce reallocations while still preserving the intended geometric growth behavior (and it would also align with the PR description’s mention of a per-chunk capacity hint).

                pastEvents = new List<HistoryEvent>();
                await foreach (P.HistoryChunk chunk in streamResponse.ResponseStream.ReadAllAsync(cancellation))
                {
                    foreach (P.HistoryEvent protoEvent in chunk.Events)
                    {