Performance: materialize streamed orchestration history incrementally

[berndverst]

1. What is the issue?

When a work item requires streamed history, the worker accumulates chunks by repeatedly chaining lazy Concat and Select sequences, then materializes the entire chain at the end.

2. Likely priority

Medium. It affects only orchestrations whose history is large enough to require server-side streaming, but those are the workloads where replay efficiency matters most.

3. Impact on performance

The current path creates one or more deferred iterator wrappers per history chunk. Materialization later walks the nested chain and can add allocation and enumeration overhead that grows with chunk count; older target runtimes are more susceptible to poor multi-concat behavior.

4. Details and code reference

Relevant runtime-state construction:

durabletask-dotnet/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs

Lines 250 to 292 in 883211a

	async ValueTask<OrchestrationRuntimeState> BuildRuntimeStateAsync(
	P.OrchestratorRequest orchestratorRequest,
	ProtoUtils.EntityConversionState? entityConversionState,
	CancellationToken cancellation)
	{
	Func<P.HistoryEvent, HistoryEvent> converter = entityConversionState is null
	? ProtoUtils.ConvertHistoryEvent
	: entityConversionState.ConvertFromProto;
	IEnumerable<HistoryEvent> pastEvents = [];
	if (orchestratorRequest.RequiresHistoryStreaming)
	{
	// Stream the remaining events from the remote service
	P.StreamInstanceHistoryRequest streamRequest = new()
	{
	InstanceId = orchestratorRequest.InstanceId,
	ExecutionId = orchestratorRequest.ExecutionId,
	ForWorkItemProcessing = true,
	};
	using AsyncServerStreamingCall<P.HistoryChunk> streamResponse =
	this.client.StreamInstanceHistory(streamRequest, cancellationToken: cancellation);
	await foreach (P.HistoryChunk chunk in streamResponse.ResponseStream.ReadAllAsync(cancellation))
	{
	pastEvents = pastEvents.Concat(chunk.Events.Select(converter));
	}
	}
	else
	{
	// The history was already provided in the work item request
	pastEvents = orchestratorRequest.PastEvents.Select(converter);
	}
	IEnumerable<HistoryEvent> 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)
	{
	// AddEvent() puts events into the NewEvents list.
	runtimeState.AddEvent(e);
	}

BuildRuntimeStateAsync initializes an IEnumerable<HistoryEvent>, extends it with pastEvents.Concat(chunk.Events.Select(converter)) for each streamed chunk, and finally calls pastEvents.ToList().

5. Guidance for how to fix

Use one List<HistoryEvent> for streamed history and call AddRange(chunk.Events.Select(converter)) as each chunk arrives. Pass that list directly to OrchestrationRuntimeState. Preserve event order exactly, and add a test covering multiple chunks and entity-event conversion to protect replay behavior.