Performance: reduce orchestration history replay allocations

Long orchestration histories were copied and rescanned several times per
work item before and during replay:

- `TaskHubGrpcWorker._execute_orchestrator()` built
  `list(req.pastEvents) + list(req.newEvents)` (three throwaway lists sized
  by the full history) purely to locate the first `executionStarted` event.
  It now iterates `itertools.chain(...)` and still breaks at the first match.
- `_OrchestrationExecutor.execute()` materialized every `executionStarted`
  event in the committed history but only read element `[0]`. It now uses
  `next(...)` over a generator, short-circuiting at the first match and
  falling back to `"<unknown>"` exactly as before.
- `_OrchestrationExecutor._build_rewind_result()` built a combined
  `list(old_events) + list(new_events)` for its two passes. Both inputs are
  sequences, so each pass now chains them lazily in the same order.

Replay order, rewind handling, non-determinism detection, log messages and
exception types/messages are all unchanged; this is an allocation-only
change with no observable behavior difference.

Adds regression tests covering both branches of the orchestration-name
lookup (resolved from committed history, and the "<unknown>" fallback).

Fixes #185

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: d6a6454c-5bdc-4bc7-9059-a2789a65e271

Author: Bernd Verst

durabletask/worker.py

@@ -3,6 +3,7 @@
 
 import asyncio
 import inspect
+import itertools
 import json
 import logging
 import os
@@ -1132,10 +1133,12 @@ def _execute_orchestrator(
         if self._payload_store is not None:
             payload_helpers.deexternalize_payloads(req, self._payload_store)
 
-        # Extract parent trace context from executionStarted event
+        # Extract parent trace context from executionStarted event.
+        # Chain the two event lists lazily so that long histories are not
+        # copied just to locate the first executionStarted event.
         parent_trace_ctx = None
         orchestration_name = "<unknown>"
-        for e in list(req.pastEvents) + list(req.newEvents):
+        for e in itertools.chain(req.pastEvents, req.newEvents):
             if e.HasField("executionStarted"):
                 orchestration_name = e.executionStarted.name
                 if e.executionStarted.HasField("parentTraceContext"):
@@ -2132,10 +2135,12 @@ def execute(
             old_events: Sequence[pb.HistoryEvent],
             new_events: Sequence[pb.HistoryEvent],
     ) -> ExecutionResults:
-        orchestration_name = "<unknown>"
-        orchestration_started_events = [e for e in old_events if e.HasField("executionStarted")]
-        if len(orchestration_started_events) >= 1:
-            orchestration_name = orchestration_started_events[0].executionStarted.name
+        # Only the first executionStarted event is needed, so stop at the
+        # first match instead of materializing every matching event.
+        orchestration_name = next(
+            (e.executionStarted.name for e in old_events if e.HasField("executionStarted")),
+            "<unknown>",
+        )
         self._orchestration_name = orchestration_name
 
         self._logger.debug(
@@ -2286,21 +2291,22 @@ def _build_rewind_result(
 
         rewind_event: pb.ExecutionRewoundEvent = new_events[1].executionRewound
 
-        all_events = list(old_events) + list(new_events)
         # Generate a new execution ID for the rewound execution.
         new_execution_id = uuid.uuid4().hex
 
         # First pass: collect the task-scheduled IDs that correspond to
         # failed activities so we can remove the matching taskScheduled
-        # events in the second pass.
+        # events in the second pass.  Both arguments are sequences, so each
+        # pass chains them lazily (old events first, then new events)
+        # instead of copying the whole history into a combined list.
         failed_task_ids: set[int] = set()
-        for event in all_events:
+        for event in itertools.chain(old_events, new_events):
             if event.HasField("taskFailed"):
                 failed_task_ids.add(event.taskFailed.taskScheduledId)
 
         # Second pass: build the clean history.
         clean_history: list[pb.HistoryEvent] = []
-        for event in all_events:
+        for event in itertools.chain(old_events, new_events):
             if event.HasField("taskFailed"):
                 continue
             if event.HasField("taskScheduled") and event.eventId in failed_task_ids:

tests/durabletask/test_orchestration_executor.py

@@ -93,6 +93,49 @@ def orchestrator(ctx: task.OrchestrationContext, _):
     assert observed["parent"] is None
 
 
+def test_orchestration_name_resolved_from_committed_history():
+    """The orchestration name is taken from the first executionStarted event in old events."""
+
+    def dummy_activity(ctx, _):
+        pass
+
+    def orchestrator(ctx: task.OrchestrationContext, _):
+        result = yield ctx.call_activity(dummy_activity, input=None)
+        return result
+
+    registry = worker._Registry()
+    name = registry.add_orchestrator(orchestrator)
+
+    # executionStarted is not the first event, so the whole prefix of the
+    # committed history has to be scanned before the name is found.
+    old_events = [
+        helpers.new_orchestrator_started_event(),
+        helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=None),
+        helpers.new_task_scheduled_event(1, task.get_name(dummy_activity))]
+    new_events = [helpers.new_task_completed_event(1, json.dumps("done!"))]
+
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter())
+    executor.execute(TEST_INSTANCE_ID, old_events, new_events)
+
+    assert executor._orchestration_name == name
+
+
+def test_orchestration_name_unknown_without_committed_history():
+    """The orchestration name falls back to '<unknown>' when old events carry no executionStarted."""
+
+    def orchestrator(ctx: task.OrchestrationContext, _):
+        return "done"
+
+    registry = worker._Registry()
+    name = registry.add_orchestrator(orchestrator)
+
+    new_events = [helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=None)]
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter())
+    executor.execute(TEST_INSTANCE_ID, [], new_events)
+
+    assert executor._orchestration_name == "<unknown>"
+
+
 def test_complete_orchestration_actions():
     """Tests the actions output for a completed orchestration"""