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"""