From 8ea44a26b022c43794877fb0419650fb5a97f289 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Sun, 26 Jul 2026 23:12:38 -0700 Subject: [PATCH] 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 `""` 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 "" fallback). Fixes #185 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: d6a6454c-5bdc-4bc7-9059-a2789a65e271 --- durabletask/worker.py | 26 ++++++----- .../test_orchestration_executor.py | 43 +++++++++++++++++++ 2 files changed, 59 insertions(+), 10 deletions(-) diff --git a/durabletask/worker.py b/durabletask/worker.py index 7efb732c..1770b758 100644 --- a/durabletask/worker.py +++ b/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 = "" - 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 = "" - 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")), + "", + ) 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: diff --git a/tests/durabletask/test_orchestration_executor.py b/tests/durabletask/test_orchestration_executor.py index 3df00a76..0b845671 100644 --- a/tests/durabletask/test_orchestration_executor.py +++ b/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 '' 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 == "" + + def test_complete_orchestration_actions(): """Tests the actions output for a completed orchestration"""