diff --git a/CHANGELOG.md b/CHANGELOG.md index 31036bfed..9da68df8e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,15 @@ to include examples, links to docs, or any other relevant information. ### Changed +- Prepared replay-safe workflow activation scheduling that prevents cancellation + from being lost when another event becomes ready in the same workflow task. The + behavior is guarded by internal workflow logic flag 2 and remains disabled by + default during its compatibility rollout. + **Maintainer reminder:** keep flag 2 default-disabled for the first two published + SDK releases that recognize it; enable it in the third release, remove the explicit + overrides for this flag from `tests/worker/test_workflow.py`, and replace this rollout + note with a `Fixed` entry announcing the behavior change. + ### Deprecated ### Breaking Changes diff --git a/temporalio/worker/_replayer.py b/temporalio/worker/_replayer.py index 61dcb84f4..b499b428b 100644 --- a/temporalio/worker/_replayer.py +++ b/temporalio/worker/_replayer.py @@ -24,7 +24,12 @@ from ._interceptor import Interceptor from ._worker import load_default_build_id from ._workflow import _WorkflowWorker -from ._workflow_instance import UnsandboxedWorkflowRunner, WorkflowRunner +from ._workflow_instance import ( + _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS, + UnsandboxedWorkflowRunner, + WorkflowRunner, + _WorkflowLogicFlag, +) from .workflow_sandbox import SandboxedWorkflowRunner logger = logging.getLogger(__name__) @@ -83,6 +88,7 @@ def __init__( header_codec_behavior=header_codec_behavior, ) self._initial_config = self._config.copy() + self._default_workflow_logic_flags = set(_DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS) # Apply plugin configuration self.plugins = plugins @@ -93,6 +99,14 @@ def __init__( if not self._config.get("workflows"): raise ValueError("At least one workflow must be specified") + def _set_default_workflow_logic_flag( + self, flag: _WorkflowLogicFlag, *, enabled: bool + ) -> None: + if enabled: + self._default_workflow_logic_flags.add(flag) + else: + self._default_workflow_logic_flags.discard(flag) + def config(self, *, active_config: bool = False) -> ReplayerConfig: """Config, as a dictionary, used to create this replayer. @@ -270,6 +284,9 @@ def on_eviction_hook( ) != HeaderCodecBehavior.NO_CODEC, max_workflow_task_external_storage_concurrency=1, + default_workflow_logic_flags=frozenset( + self._default_workflow_logic_flags + ), ) external_storage = data_converter.external_storage storage_driver_types = ( diff --git a/temporalio/worker/_worker.py b/temporalio/worker/_worker.py index 5e2d8ce58..b48a27918 100644 --- a/temporalio/worker/_worker.py +++ b/temporalio/worker/_worker.py @@ -43,6 +43,7 @@ PatchActivationInput, UnsandboxedWorkflowRunner, WorkflowRunner, + _WorkflowLogicFlag, ) from .workflow_sandbox import SandboxedWorkflowRunner @@ -735,6 +736,17 @@ def client(self, value: temporalio.client.Client) -> None: if self._nexus_worker: self._nexus_worker._client = value + def _set_default_workflow_logic_flag( + self, flag: _WorkflowLogicFlag, *, enabled: bool + ) -> None: + if self._started: + raise RuntimeError( + "Cannot set default workflow logic flags after the worker has started" + ) + if not self._workflow_worker: + raise RuntimeError("Cannot set workflow logic flags without workflows") + self._workflow_worker._set_default_workflow_logic_flag(flag, enabled=enabled) + @property def is_running(self) -> bool: """Whether the worker is running. diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py index b9513068d..853483b55 100644 --- a/temporalio/worker/_workflow.py +++ b/temporalio/worker/_workflow.py @@ -40,11 +40,13 @@ WorkflowInterceptorClassInput, ) from ._workflow_instance import ( + _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS, PatchActivationInput, WorkflowInstance, WorkflowInstanceDetails, WorkflowRunner, _WorkflowExternFunctions, + _WorkflowLogicFlag, ) logger = logging.getLogger(__name__) @@ -90,6 +92,7 @@ def __init__( assert_local_activity_valid: Callable[[str], None], encode_headers: bool, max_workflow_task_external_storage_concurrency: int, + default_workflow_logic_flags: frozenset[_WorkflowLogicFlag] | None = None, ) -> None: # Debug mode is enabled if specified or if the TEMPORAL_DEBUG env var is truthy debug_mode = debug_mode or bool(os.environ.get("TEMPORAL_DEBUG")) @@ -97,6 +100,11 @@ def __init__( self._bridge_worker = bridge_worker self._namespace = namespace self._task_queue = task_queue + self._default_workflow_logic_flags = set( + _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS + if default_workflow_logic_flags is None + else default_workflow_logic_flags + ) self._workflow_task_executor = ( workflow_task_executor or concurrent.futures.ThreadPoolExecutor( @@ -788,6 +796,7 @@ def _create_workflow_instance( patch_activation_callback=self._patch_activation_callback, last_completion_result=init.last_completion_result, last_failure=last_failure, + default_workflow_logic_flags=frozenset(self._default_workflow_logic_flags), ) if defn.sandboxed: return self._workflow_runner.create_instance(det) @@ -800,6 +809,14 @@ def nondeterminism_as_workflow_fail(self) -> bool: for typ in self._workflow_failure_exception_types ) + def _set_default_workflow_logic_flag( + self, flag: _WorkflowLogicFlag, *, enabled: bool + ) -> None: + if enabled: + self._default_workflow_logic_flags.add(flag) + else: + self._default_workflow_logic_flags.discard(flag) + def nondeterminism_as_workflow_fail_for_types(self) -> set[str]: return { k diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 726ff85e0..1e680a78e 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -27,7 +27,7 @@ Sequence, ) from contextlib import contextmanager -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import timedelta from enum import IntEnum from typing import ( @@ -169,6 +169,9 @@ class WorkflowInstanceDetails: patch_activation_callback: Callable[[PatchActivationInput], bool] | None last_completion_result: temporalio.api.common.v1.Payloads last_failure: Failure | None + default_workflow_logic_flags: frozenset[_WorkflowLogicFlag] = field( + default_factory=lambda: _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS + ) class WorkflowInstance(ABC): @@ -288,7 +291,9 @@ def __init__(self, det: WorkflowInstanceDetails) -> None: det.worker_level_failure_exception_types ) self._patch_activation_callback = det.patch_activation_callback + self._default_workflow_logic_flags = det.default_workflow_logic_flags self._primary_task: asyncio.Task[None] | None = None + self._cancel_primary_task_pending = False self._time_ns = 0 self._cancel_reason: str | None = None self._deployment_version_for_current_task: None | ( @@ -459,6 +464,9 @@ def activate( self._is_replaying = act.is_replaying self._current_thread_id = threading.get_ident() self._current_internal_flags = act.available_internal_flags + self._single_batch_activation = self._workflow_logic_flag_enabled( + _WorkflowLogicFlag.PROCESS_WORKFLOW_ACTIVATION_JOBS_AS_SINGLE_BATCH + ) activation_err: Exception | None = None try: # Split into job sets with patches, then signals + updates, then @@ -479,21 +487,39 @@ def activate( else: job_sets[3].append(job) + # Core guarantees query-only activations. Fail the workflow task if violated. + assert not job_sets[3] or not any(job_sets[:3]), ( + "Query jobs must not share an activation with non-query jobs. " + "This is an SDK Core bug." + ) + if start_job: self._workflow_input = self._make_workflow_input(start_job) - # Apply every job set, running after each set - for index, job_set in enumerate(job_sets): - if not job_set: - continue - for job in job_set: - # Let errors bubble out of these to the caller to fail the task - self._apply(job) - - # Run one iteration of the loop. We do not allow conditions to - # be checked in patch jobs (first index) or query jobs (last - # index). - self._run_once(check_conditions=index == 1 or index == 2) + if self._single_batch_activation: + # Applying every job before giving workflow tasks a chance to + # run prevents their order in the activation from hiding state + # that arrived in the same workflow task. + for job_set in job_sets: + for job in job_set: + # Let errors bubble out of these to the caller to fail the task + self._apply(job) + if any(job_sets): + self._run_once(check_conditions=bool(job_sets[1] or job_sets[2])) + else: + # Preserve the legacy scheduling order for histories which do + # not contain the single-batch workflow logic flag. + for index, job_set in enumerate(job_sets): + if not job_set: + continue + for job in job_set: + # Let errors bubble out of these to the caller to fail the task + self._apply(job) + + # Run one iteration of the loop. We do not allow conditions to + # be checked in patch jobs (first index) or query jobs (last + # index). + self._run_once(check_conditions=index == 1 or index == 2) except Exception as err: # We want some errors during activation, like those that can happen # during payload conversion, to be able to fail the workflow not the @@ -628,6 +654,10 @@ def _apply_cancel_workflow( # workflow the ability to receive the cancellation, so we must defer # this cancellation to the next iteration of the event loop. self.call_soon(self._primary_task.cancel) + elif self._single_batch_activation: + # Initialization is the only job that creates the primary task, so + # retain a same-activation cancellation until that task exists. + self._cancel_primary_task_pending = True def _apply_do_update( self, job: temporalio.bridge.proto.workflow_activation.DoUpdate @@ -1121,6 +1151,9 @@ async def run_workflow(input: ExecuteWorkflowInput) -> None: self._run_top_level_workflow_function(run_workflow(self._workflow_input)), name="run", ) + if self._cancel_primary_task_pending: + self._cancel_primary_task_pending = False + self.call_soon(self._primary_task.cancel) def _apply_update_random_seed( self, job: temporalio.bridge.proto.workflow_activation.UpdateRandomSeed @@ -1832,6 +1865,19 @@ async def workflow_wait_condition( timeout_summary: str | None = None, ) -> None: self._assert_not_read_only("wait condition") + cancellation_requested_before = self._cancel_reason is not None + + # Some asyncio.wait_for implementations can prefer a ready condition + # result or timeout over task cancellation that becomes ready in the same + # event-loop turn. Only detect a new request so workflows can catch + # cancellation and keep going. + def cancellation_arrived() -> bool: + return ( + self._single_batch_activation + and not cancellation_requested_before + and self._cancel_reason is not None + ) + fut = self.create_future() self._conditions.append((fn, fut)) user_metadata = ( @@ -1849,7 +1895,14 @@ async def in_context(): _TimerOptionsCtxVar.set(_TimerOptions(user_metadata=user_metadata)) await asyncio.wait_for(fut, timeout) - await ctxvars.run(in_context) + try: + await ctxvars.run(in_context) + except asyncio.TimeoutError: + if cancellation_arrived(): + raise asyncio.CancelledError() + raise + if cancellation_arrived(): + raise asyncio.CancelledError() def workflow_get_current_details(self) -> str: return self._current_details @@ -1938,10 +1991,11 @@ async def run_activity() -> Any: # be marked as unstarted handle._started = True try: - # We use _shield_await instead of asyncio.shield to prevent - # the underlying result future from being cancelled while avoiding - # a spurious error log on Python 3.11+ (see issue #1600). - return await _shield_await(handle._result_fut) + return await self._await_temporal_operation( + handle._result_fut, + lambda _err, command: handle._apply_cancel_command(command), + completed_cancellation_flag=_WorkflowLogicFlag.RAISE_ON_CANCELLING_COMPLETED_ACTIVITY, + ) except _ActivityDoBackoffError as err: # We have to sleep then reschedule. Note this sleep can be # cancelled like any other timer. @@ -1952,28 +2006,6 @@ async def run_activity() -> Any: # We have to put the handle back on the pending activity # dict with its new seq self._pending_activities[handle._seq] = handle - except asyncio.CancelledError: - # If an activity future completes at the same time as a cancellation is being processed, the cancellation would be swallowed - # _WorkflowLogicFlag.RAISE_ON_CANCELLING_COMPLETED_ACTIVITY will correctly reraise the exception - if handle._result_fut.done(): - if ( - not self._is_replaying - or _WorkflowLogicFlag.RAISE_ON_CANCELLING_COMPLETED_ACTIVITY - in self._current_internal_flags - ): - self._current_completion.successful.used_internal_flags.append( - _WorkflowLogicFlag.RAISE_ON_CANCELLING_COMPLETED_ACTIVITY - ) - raise - # Send a cancel request to the activity - handle._apply_cancel_command(self._add_command()) - # Clear the cancellation counter on Python 3.11+ so the - # next await does not immediately re-raise CancelledError - if ( - sys.version_info >= (3, 11) - and (t := asyncio.current_task()) is not None - ): - t.uncancel() # type: ignore[union-attr] # Create the handle and set as pending handle = _ActivityHandle(self, input, run_activity()) @@ -2030,11 +2062,13 @@ async def _outbound_start_child_workflow( handle: _ChildWorkflowHandle # Common code for handling cancel for start and run - def apply_child_cancel_error(err: asyncio.CancelledError) -> None: + def apply_child_cancel_error( + err: asyncio.CancelledError, + cancel_command: temporalio.bridge.proto.workflow_commands.WorkflowCommand, + ) -> None: # Send a cancel request to the child, forwarding the msg passed to # Task.cancel(msg) (if any) as the cancellation reason. reason = err.args[0] if err.args and isinstance(err.args[0], str) else "" - cancel_command = self._add_command() handle._apply_cancel_command(cancel_command, reason=reason) # If the cancel command is for external workflow, we # have to add a seq and mark it pending @@ -2053,21 +2087,9 @@ def apply_child_cancel_error(err: asyncio.CancelledError) -> None: # Function that runs in the handle async def run_child() -> Any: - while True: - try: - # We use _shield_await instead of asyncio.shield to prevent - # the future itself from being cancelled while avoiding a - # spurious error log on Python 3.11+ (see issue #1600). - return await _shield_await(handle._result_fut) - except asyncio.CancelledError as err: - apply_child_cancel_error(err) - # Clear the cancellation counter on Python 3.11+ so the - # next await does not immediately re-raise CancelledError - if ( - sys.version_info >= (3, 11) - and (t := asyncio.current_task()) is not None - ): - t.uncancel() # type: ignore[union-attr] + return await self._await_temporal_operation( + handle._result_fut, apply_child_cancel_error + ) # Create the handle and set as pending handle = _ChildWorkflowHandle( @@ -2077,24 +2099,12 @@ async def run_child() -> Any: self._pending_child_workflows[handle._seq] = handle # Wait on start before returning - while True: - try: - # We use _shield_await instead of asyncio.shield to prevent - # the future itself from being cancelled while avoiding a - # spurious error log on Python 3.11+ (see issue #1600). - await _shield_await(handle._start_fut) - return handle - except asyncio.CancelledError as err: - apply_child_cancel_error(err) - # Clear the cancellation counter on Python 3.11+ so the - # next await does not immediately re-raise CancelledError - if ( - sys.version_info >= (3, 11) - and (t := asyncio.current_task()) is not None - ): - t.uncancel() # type: ignore[union-attr] - if self._cancel_reason is not None or self._deleting: - raise + await self._await_temporal_operation( + handle._start_fut, + apply_child_cancel_error, + reraise_on_workflow_cancellation=True, + ) + return handle async def _outbound_start_nexus_operation( self, input: StartNexusOperationInput[Any, OutputT] @@ -2115,19 +2125,13 @@ async def _outbound_start_nexus_operation( handle: _NexusOperationHandle[OutputT] async def operation_handle_fn() -> OutputT: - while True: - try: - return cast(OutputT, await _shield_await(handle._result_fut)) - except asyncio.CancelledError: - cancel_command = self._add_command() - handle._apply_cancel_command(cancel_command) - # Clear the cancellation counter on Python 3.11+ so the - # next await does not immediately re-raise CancelledError - if ( - sys.version_info >= (3, 11) - and (t := asyncio.current_task()) is not None - ): - t.uncancel() # type: ignore[union-attr] + return cast( + OutputT, + await self._await_temporal_operation( + handle._result_fut, + lambda _err, command: handle._apply_cancel_command(command), + ), + ) payload_converter = ( temporalio.nexus.system.get_payload_converter() @@ -2144,25 +2148,12 @@ async def operation_handle_fn() -> OutputT: handle._apply_schedule_command() self._pending_nexus_operations[handle._seq] = handle - while True: - try: - # We use _shield_await instead of asyncio.shield to prevent - # the future itself from being cancelled while avoiding a - # spurious error log on Python 3.11+ (see issue #1600). - await _shield_await(handle._start_fut) - return handle - except asyncio.CancelledError: - cancel_command = self._add_command() - handle._apply_cancel_command(cancel_command) - # Clear the cancellation counter on Python 3.11+ so the - # next await does not immediately re-raise CancelledError - if ( - sys.version_info >= (3, 11) - and (t := asyncio.current_task()) is not None - ): - t.uncancel() # type: ignore[union-attr] - if self._cancel_reason is not None or self._deleting: - raise + await self._await_temporal_operation( + handle._start_fut, + lambda _err, command: handle._apply_cancel_command(command), + reraise_on_workflow_cancellation=True, + ) + return handle #### Miscellaneous helpers #### # These are in alphabetical order. @@ -2171,6 +2162,14 @@ def _add_command(self) -> temporalio.bridge.proto.workflow_commands.WorkflowComm self._assert_not_read_only("add command") return self._current_completion.successful.commands.add() + def _workflow_logic_flag_enabled(self, flag: _WorkflowLogicFlag) -> bool: + if flag in self._current_internal_flags: + return True + if self._is_replaying or flag not in self._default_workflow_logic_flags: + return False + self._current_completion.successful.used_internal_flags.append(flag) + return True + @contextmanager def _as_read_only(self, *, in_query_or_validator: bool) -> Iterator[None]: prev_read_only = self._read_only @@ -2195,6 +2194,56 @@ def _assert_not_read_only( f"While in read-only function, action attempted: {action_attempted}" ) + async def _await_temporal_operation( + self, + fut: asyncio.Future[_T], + apply_cancel: Callable[ + [ + asyncio.CancelledError, + temporalio.bridge.proto.workflow_commands.WorkflowCommand, + ], + None, + ], + *, + completed_cancellation_flag: _WorkflowLogicFlag | None = None, + reraise_on_workflow_cancellation: bool = False, + ) -> _T: + while True: + try: + # Protect the operation's result from task cancellation so a + # Temporal cancellation command can decide its outcome. The + # custom shield also avoids spurious error logs on Python 3.11+. + return await _shield_await(fut) + except asyncio.CancelledError as err: + if fut.done(): + # Retrying the shield after both futures become ready would + # return the result and erase the task cancellation. + if self._single_batch_activation: + raise + if completed_cancellation_flag is not None and ( + not self._is_replaying + or completed_cancellation_flag in self._current_internal_flags + ): + self._current_completion.successful.used_internal_flags.append( + completed_cancellation_flag + ) + raise + + apply_cancel(err, self._add_command()) + + # Clear the cancellation counter on Python 3.11+ so the next + # await does not immediately re-raise CancelledError. + if ( + sys.version_info >= (3, 11) + and (task := asyncio.current_task()) is not None + ): + task.uncancel() # type: ignore[union-attr] + + if reraise_on_workflow_cancellation and ( + self._cancel_reason is not None or self._deleting + ): + raise + async def _cancel_external_workflow( self, # Should not have seq set @@ -2685,23 +2734,14 @@ async def _signal_external_workflow( ) self._pending_external_signals[seq] = (done_fut, target_workflow_id) + def apply_cancel( + _err: asyncio.CancelledError, + command: temporalio.bridge.proto.workflow_commands.WorkflowCommand, + ) -> None: + command.cancel_signal_workflow.seq = seq + # Wait until completed or cancelled - while True: - try: - # We use _shield_await instead of asyncio.shield to prevent - # the future itself from being cancelled while avoiding a - # spurious error log on Python 3.11+ (see issue #1600). - return await _shield_await(done_fut) - except asyncio.CancelledError: - cancel_command = self._add_command() - cancel_command.cancel_signal_workflow.seq = seq - # Clear the cancellation counter on Python 3.11+ so the - # next await does not immediately re-raise CancelledError - if ( - sys.version_info >= (3, 11) - and (t := asyncio.current_task()) is not None - ): - t.uncancel() # type: ignore[union-attr] + return await self._await_temporal_operation(done_fut, apply_cancel) def _stack_trace(self) -> str: stacks = [] @@ -3894,3 +3934,11 @@ class _WorkflowLogicFlag(IntEnum): """Flags that may be set on task/activation completion to differentiate new from old workflow behavior.""" RAISE_ON_CANCELLING_COMPLETED_ACTIVITY = 1 + PROCESS_WORKFLOW_ACTIVATION_JOBS_AS_SINGLE_BATCH = 2 + + +# TODO: Enable PROCESS_WORKFLOW_ACTIVATION_JOBS_AS_SINGLE_BATCH by default after +# two published SDK releases have recognized flag 2. When enabling it, remove the +# explicit overrides for this flag from tests/worker/test_workflow.py, then remove +# this reminder. +_DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS: frozenset[_WorkflowLogicFlag] = frozenset() diff --git a/tests/worker/test_replayer.py b/tests/worker/test_replayer.py index 22d771556..137b32c3e 100644 --- a/tests/worker/test_replayer.py +++ b/tests/worker/test_replayer.py @@ -10,6 +10,7 @@ import pytest +import temporalio.worker._workflow_instance from temporalio import activity, workflow from temporalio.client import Client, WorkflowFailureError, WorkflowHistory from temporalio.exceptions import ApplicationError @@ -81,6 +82,24 @@ def waiting(self) -> bool: return self._waiting +_WorkflowLogicFlag = temporalio.worker._workflow_instance._WorkflowLogicFlag +_SINGLE_BATCH_WORKFLOW_LOGIC_FLAG = ( + _WorkflowLogicFlag.PROCESS_WORKFLOW_ACTIVATION_JOBS_AS_SINGLE_BATCH +) + + +def _history_uses_workflow_logic_flag( + history: WorkflowHistory, + flag: _WorkflowLogicFlag, +) -> bool: + return any( + event.HasField("workflow_task_completed_event_attributes") + and int(flag) + in event.workflow_task_completed_event_attributes.sdk_metadata.lang_used_flags + for event in history.events + ) + + @pytest.mark.skipif(sys.version_info < (3, 12), reason="Skipping for < 3.12") async def test_replayer_workflow_complete(client: Client) -> None: # This test skips for versions < 3.12 because this is flaky due to CPython reimport issue: @@ -427,15 +446,12 @@ async def execute_workflow(self, input: ExecuteWorkflowInput) -> Any: return res -async def test_replayer_async_ordering() -> None: - """ - This test verifies that the order that asyncio tasks/coroutines are woken up matches the - order they were before changes to apply all jobs and then run the event loop, where previously - the event loop was ran after each "batch" of jobs. - """ - histories_and_expecteds = [ - ( +@pytest.mark.parametrize( + ("history_filename", "uses_single_batch", "expected"), + [ + pytest.param( "test_replayer_event_tracing.json", + False, [ "sig-before-sync", "sig-before-1", @@ -453,9 +469,33 @@ async def test_replayer_async_ordering() -> None: "timer-1", "timer-2", ], + id="legacy-event-tracing", ), - ( + pytest.param( + "test_replayer_event_tracing_single_batch.json", + True, + [ + "sig-before-sync", + "sig-before-1", + "timer-sync", + "act-sync", + "sig-before-2", + "act-1", + "act-2", + "sig-1-sync", + "sig-1-1", + "sig-1-2", + "update-1-sync", + "update-1-1", + "update-1-2", + "timer-1", + "timer-2", + ], + id="single-batch-event-tracing", + ), + pytest.param( "test_replayer_event_tracing_double_sig_at_start.json", + False, [ "sig-before-sync", "sig-before-1", @@ -473,29 +513,72 @@ async def test_replayer_async_ordering() -> None: "timer-1", "timer-2", ], + id="legacy-double-signal-at-start", ), - ] - for history, expected in histories_and_expecteds: - with Path(__file__).with_name(history).open() as f: - history = f.read() - await Replayer( - workflows=[SignalsActivitiesTimersUpdatesTracingWorkflow], - interceptors=[WorkerWorkflowResultInterceptor()], - ).replay_workflow(WorkflowHistory.from_json("fake", history)) - assert test_replayer_workflow_res == expected + pytest.param( + "test_replayer_event_tracing_double_sig_at_start_single_batch.json", + True, + [ + "sig-before-sync", + "sig-before-1", + "sig-1-sync", + "sig-1-1", + "timer-sync", + "act-sync", + "sig-before-2", + "sig-1-2", + "act-1", + "act-2", + "update-1-sync", + "update-1-1", + "update-1-2", + "timer-1", + "timer-2", + ], + id="single-batch-double-signal-at-start", + ), + ], +) +async def test_replayer_async_ordering( + history_filename: str, + uses_single_batch: bool, + expected: list[str], +) -> None: + """ + Verify legacy and single-batch histories replay with the asyncio scheduling order that their + original executions observed. + """ + with Path(__file__).with_name(history_filename).open() as f: + history = WorkflowHistory.from_json("fake", f.read()) + assert ( + _history_uses_workflow_logic_flag(history, _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG) + is uses_single_batch + ) + await Replayer( + workflows=[SignalsActivitiesTimersUpdatesTracingWorkflow], + interceptors=[WorkerWorkflowResultInterceptor()], + ).replay_workflow(history) + assert test_replayer_workflow_res == expected -async def test_replayer_alternate_async_ordering() -> None: +async def test_replayer_unflagged_history_uses_legacy_async_ordering() -> None: with ( Path(__file__) .with_name("test_replayer_event_tracing_alternate.json") .open() as f ): - history = f.read() - await Replayer( + history = WorkflowHistory.from_json("fake", f.read()) + assert not _history_uses_workflow_logic_flag( + history, _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG + ) + replayer = Replayer( workflows=[ActivityAndSignalsWhileWorkflowDown], interceptors=[WorkerWorkflowResultInterceptor()], - ).replay_workflow(WorkflowHistory.from_json("fake", history)) + ) + replayer._set_default_workflow_logic_flag( + _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG, enabled=True + ) + await replayer.replay_workflow(history) assert test_replayer_workflow_res == [ "act-start", "sig-1", diff --git a/tests/worker/test_replayer_event_tracing_double_sig_at_start_single_batch.json b/tests/worker/test_replayer_event_tracing_double_sig_at_start_single_batch.json new file mode 100644 index 000000000..ed0eae143 --- /dev/null +++ b/tests/worker/test_replayer_event_tracing_double_sig_at_start_single_batch.json @@ -0,0 +1,433 @@ +{ + "events": [ + { + "eventId": "1", + "eventTime": "2026-07-22T17:46:09.820508351Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_STARTED", + "taskId": "1048655", + "workflowExecutionStartedEventAttributes": { + "workflowType": { + "name": "SignalsActivitiesTimersUpdatesTracingWorkflow" + }, + "taskQueue": { + "name": "tq-104f9b8b-4835-4705-8f36-3fb3f0b9bce7", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "workflowTaskTimeout": "10s", + "originalExecutionRunId": "019f8aef-5a1c-77c0-8c6f-4c42b8984e71", + "identity": "2468547@monolith", + "firstExecutionRunId": "019f8aef-5a1c-77c0-8c6f-4c42b8984e71", + "attempt": 1, + "firstWorkflowTaskBackoff": "0s", + "workflowId": "wf-50105b28-aad3-488b-b8b5-6cd3c168369c", + "priority": {} + } + }, + { + "eventId": "2", + "eventTime": "2026-07-22T17:46:09.820553958Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048656", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "tq-104f9b8b-4835-4705-8f36-3fb3f0b9bce7", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "3", + "eventTime": "2026-07-22T17:46:09.823263829Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED", + "taskId": "1048661", + "workflowExecutionSignaledEventAttributes": { + "signalName": "dosig", + "input": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "ImJlZm9yZSI=" + } + ] + }, + "identity": "2468547@monolith", + "requestId": "0f1accfe-ffda-4af1-bc26-2a4b43007ef5" + } + }, + { + "eventId": "4", + "eventTime": "2026-07-22T17:46:09.826228197Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED", + "taskId": "1048663", + "workflowExecutionSignaledEventAttributes": { + "signalName": "dosig", + "input": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "IjEi" + } + ] + }, + "identity": "2468547@monolith", + "requestId": "7c76e2c6-c5c9-4bfa-9ef8-680e5fa0fdac" + } + }, + { + "eventId": "5", + "eventTime": "2026-07-22T17:46:09.911987108Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048665", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "2", + "identity": "2468547@monolith", + "requestId": "eb18fba3-0020-4bee-9824-68dee535759c", + "historySizeBytes": "600", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + } + } + }, + { + "eventId": "6", + "eventTime": "2026-07-22T17:46:10.029696095Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048669", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "2", + "startedEventId": "5", + "identity": "2468547@monolith", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + }, + "sdkMetadata": { + "coreUsedFlags": [ + 2, + 3, + 1 + ], + "langUsedFlags": [ + 2 + ], + "sdkName": "temporal-python", + "sdkVersion": "1.30.0" + }, + "meteringMetadata": {} + } + }, + { + "eventId": "7", + "eventTime": "2026-07-22T17:46:10.029723573Z", + "eventType": "EVENT_TYPE_TIMER_STARTED", + "taskId": "1048670", + "timerStartedEventAttributes": { + "timerId": "1", + "startToFireTimeout": "0.100s", + "workflowTaskCompletedEventId": "6" + } + }, + { + "eventId": "8", + "eventTime": "2026-07-22T17:46:10.029791694Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_SCHEDULED", + "taskId": "1048671", + "activityTaskScheduledEventAttributes": { + "activityId": "1", + "activityType": { + "name": "say_hello" + }, + "taskQueue": { + "name": "tq-104f9b8b-4835-4705-8f36-3fb3f0b9bce7", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "header": {}, + "input": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "IkVuY2hpIg==" + } + ] + }, + "scheduleToCloseTimeout": "30s", + "scheduleToStartTimeout": "30s", + "startToCloseTimeout": "30s", + "heartbeatTimeout": "0s", + "workflowTaskCompletedEventId": "6", + "retryPolicy": { + "initialInterval": "1s", + "backoffCoefficient": 2.0, + "maximumInterval": "100s" + }, + "useWorkflowBuildId": true, + "priority": {} + } + }, + { + "eventId": "9", + "eventTime": "2026-07-22T17:46:10.031520468Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_STARTED", + "taskId": "1048679", + "activityTaskStartedEventAttributes": { + "scheduledEventId": "8", + "identity": "2468547@monolith", + "requestId": "58a7bab7-a808-49ba-b906-3c07a281f078", + "attempt": 1, + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + } + } + }, + { + "eventId": "10", + "eventTime": "2026-07-22T17:46:10.034867651Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_COMPLETED", + "taskId": "1048680", + "activityTaskCompletedEventAttributes": { + "result": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "IkhlbGxvLCBFbmNoaSEi" + } + ] + }, + "scheduledEventId": "8", + "startedEventId": "9", + "identity": "2468547@monolith" + } + }, + { + "eventId": "11", + "eventTime": "2026-07-22T17:46:10.034876231Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048681", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "2468547@monolith-af9e32c231d44027a7fbb997838520bf", + "kind": "TASK_QUEUE_KIND_STICKY", + "normalName": "tq-104f9b8b-4835-4705-8f36-3fb3f0b9bce7" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "12", + "eventTime": "2026-07-22T17:46:10.035716599Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048685", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "11", + "identity": "2468547@monolith", + "requestId": "cf2bd6a2-3498-427d-920a-daf350415c70", + "historySizeBytes": "1389", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + } + } + }, + { + "eventId": "13", + "eventTime": "2026-07-22T17:46:10.039414094Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048689", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "11", + "startedEventId": "12", + "identity": "2468547@monolith", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + }, + "sdkMetadata": {}, + "meteringMetadata": {} + } + }, + { + "eventId": "14", + "eventTime": "2026-07-22T17:46:10.112263987Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048695", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "2468547@monolith-af9e32c231d44027a7fbb997838520bf", + "kind": "TASK_QUEUE_KIND_STICKY", + "normalName": "tq-104f9b8b-4835-4705-8f36-3fb3f0b9bce7" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "15", + "eventTime": "2026-07-22T17:46:10.112924510Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048696", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "14", + "identity": "2468547@monolith", + "requestId": "6e079d38-0b99-48eb-b1e9-1b8236232fb1", + "historySizeBytes": "1598", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + } + } + }, + { + "eventId": "16", + "eventTime": "2026-07-22T17:46:10.118047317Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048697", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "14", + "startedEventId": "15", + "identity": "2468547@monolith", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + }, + "sdkMetadata": {}, + "meteringMetadata": {} + } + }, + { + "eventId": "17", + "eventTime": "2026-07-22T17:46:10.118078114Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ACCEPTED", + "taskId": "1048698", + "workflowExecutionUpdateAcceptedEventAttributes": { + "protocolInstanceId": "3078b22e-fa4a-4bb3-bf21-7502464f3f8e", + "acceptedRequestMessageId": "3078b22e-fa4a-4bb3-bf21-7502464f3f8e/request", + "acceptedRequestSequencingEventId": "14", + "acceptedRequest": { + "meta": { + "updateId": "3078b22e-fa4a-4bb3-bf21-7502464f3f8e", + "identity": "2468547@monolith" + }, + "input": { + "name": "doupdate", + "args": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "IjEi" + } + ] + } + } + } + } + }, + { + "eventId": "18", + "eventTime": "2026-07-22T17:46:10.118094048Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_COMPLETED", + "taskId": "1048699", + "workflowExecutionUpdateCompletedEventAttributes": { + "meta": { + "updateId": "3078b22e-fa4a-4bb3-bf21-7502464f3f8e", + "identity": "2468547@monolith" + }, + "outcome": { + "success": { + "payloads": [ + { + "metadata": { + "encoding": "YmluYXJ5L251bGw=" + } + } + ] + } + }, + "acceptedEventId": "17" + } + }, + { + "eventId": "19", + "eventTime": "2026-07-22T17:46:10.822704848Z", + "eventType": "EVENT_TYPE_TIMER_FIRED", + "taskId": "1048702", + "timerFiredEventAttributes": { + "timerId": "1", + "startedEventId": "7" + } + }, + { + "eventId": "20", + "eventTime": "2026-07-22T17:46:10.822723284Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048703", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "2468547@monolith-af9e32c231d44027a7fbb997838520bf", + "kind": "TASK_QUEUE_KIND_STICKY", + "normalName": "tq-104f9b8b-4835-4705-8f36-3fb3f0b9bce7" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "21", + "eventTime": "2026-07-22T17:46:10.827860954Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048707", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "20", + "identity": "2468547@monolith", + "requestId": "facf828f-060f-4a8b-87cf-bdbb3a3297e9", + "historySizeBytes": "2430", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + } + } + }, + { + "eventId": "22", + "eventTime": "2026-07-22T17:46:10.843601707Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048711", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "20", + "startedEventId": "21", + "identity": "2468547@monolith", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + }, + "sdkMetadata": {}, + "meteringMetadata": {} + } + }, + { + "eventId": "23", + "eventTime": "2026-07-22T17:46:10.843661263Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED", + "taskId": "1048712", + "workflowExecutionCompletedEventAttributes": { + "result": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "WyJzaWctYmVmb3JlLXN5bmMiLCJzaWctYmVmb3JlLTEiLCJzaWctMS1zeW5jIiwic2lnLTEtMSIsInRpbWVyLXN5bmMiLCJhY3Qtc3luYyIsInNpZy1iZWZvcmUtMiIsInNpZy0xLTIiLCJhY3QtMSIsImFjdC0yIiwidXBkYXRlLTEtc3luYyIsInVwZGF0ZS0xLTEiLCJ1cGRhdGUtMS0yIiwidGltZXItMSIsInRpbWVyLTIiXQ==" + } + ] + }, + "workflowTaskCompletedEventId": "22" + } + } + ] +} + diff --git a/tests/worker/test_replayer_event_tracing_single_batch.json b/tests/worker/test_replayer_event_tracing_single_batch.json new file mode 100644 index 000000000..42bab0b4f --- /dev/null +++ b/tests/worker/test_replayer_event_tracing_single_batch.json @@ -0,0 +1,479 @@ +{ + "events": [ + { + "eventId": "1", + "eventTime": "2026-07-22T17:46:08.798556582Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_STARTED", + "taskId": "1048587", + "workflowExecutionStartedEventAttributes": { + "workflowType": { + "name": "SignalsActivitiesTimersUpdatesTracingWorkflow" + }, + "taskQueue": { + "name": "tq-7049fd3e-356e-49d5-9047-2b97b454562e", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "workflowTaskTimeout": "10s", + "originalExecutionRunId": "019f8aef-561e-787b-8565-31e78b4a889a", + "identity": "2468547@monolith", + "firstExecutionRunId": "019f8aef-561e-787b-8565-31e78b4a889a", + "attempt": 1, + "firstWorkflowTaskBackoff": "0s", + "workflowId": "wf-2dd53446-a657-4f47-9e47-a3edeb302c5f", + "priority": {} + } + }, + { + "eventId": "2", + "eventTime": "2026-07-22T17:46:08.798633092Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048588", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "tq-7049fd3e-356e-49d5-9047-2b97b454562e", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "3", + "eventTime": "2026-07-22T17:46:08.801258084Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED", + "taskId": "1048593", + "workflowExecutionSignaledEventAttributes": { + "signalName": "dosig", + "input": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "ImJlZm9yZSI=" + } + ] + }, + "identity": "2468547@monolith", + "requestId": "c78a6332-adb5-4ff1-af4b-f55e1180d4d8" + } + }, + { + "eventId": "4", + "eventTime": "2026-07-22T17:46:08.949831498Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048595", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "2", + "identity": "2468547@monolith", + "requestId": "1e30e279-6cc7-497a-89f5-3e032d169c49", + "historySizeBytes": "477", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + } + } + }, + { + "eventId": "5", + "eventTime": "2026-07-22T17:46:09.032870384Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048599", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "2", + "startedEventId": "4", + "identity": "2468547@monolith", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + }, + "sdkMetadata": { + "coreUsedFlags": [ + 3, + 1, + 2 + ], + "langUsedFlags": [ + 2 + ], + "sdkName": "temporal-python", + "sdkVersion": "1.30.0" + }, + "meteringMetadata": {} + } + }, + { + "eventId": "6", + "eventTime": "2026-07-22T17:46:09.032946093Z", + "eventType": "EVENT_TYPE_TIMER_STARTED", + "taskId": "1048600", + "timerStartedEventAttributes": { + "timerId": "1", + "startToFireTimeout": "0.100s", + "workflowTaskCompletedEventId": "5" + } + }, + { + "eventId": "7", + "eventTime": "2026-07-22T17:46:09.033008651Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_SCHEDULED", + "taskId": "1048601", + "activityTaskScheduledEventAttributes": { + "activityId": "1", + "activityType": { + "name": "say_hello" + }, + "taskQueue": { + "name": "tq-7049fd3e-356e-49d5-9047-2b97b454562e", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "header": {}, + "input": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "IkVuY2hpIg==" + } + ] + }, + "scheduleToCloseTimeout": "30s", + "scheduleToStartTimeout": "30s", + "startToCloseTimeout": "30s", + "heartbeatTimeout": "0s", + "workflowTaskCompletedEventId": "5", + "retryPolicy": { + "initialInterval": "1s", + "backoffCoefficient": 2.0, + "maximumInterval": "100s" + }, + "useWorkflowBuildId": true, + "priority": {} + } + }, + { + "eventId": "8", + "eventTime": "2026-07-22T17:46:09.035163378Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_STARTED", + "taskId": "1048609", + "activityTaskStartedEventAttributes": { + "scheduledEventId": "7", + "identity": "2468547@monolith", + "requestId": "bd46ae91-1b79-4c8b-9366-e24395d0da8b", + "attempt": 1, + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + } + } + }, + { + "eventId": "9", + "eventTime": "2026-07-22T17:46:09.039131108Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_COMPLETED", + "taskId": "1048610", + "activityTaskCompletedEventAttributes": { + "result": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "IkhlbGxvLCBFbmNoaSEi" + } + ] + }, + "scheduledEventId": "7", + "startedEventId": "8", + "identity": "2468547@monolith" + } + }, + { + "eventId": "10", + "eventTime": "2026-07-22T17:46:09.039143351Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048611", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "2468547@monolith-3a1c4b9f82644345a21e1663c393c0d0", + "kind": "TASK_QUEUE_KIND_STICKY", + "normalName": "tq-7049fd3e-356e-49d5-9047-2b97b454562e" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "11", + "eventTime": "2026-07-22T17:46:09.040109929Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048615", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "10", + "identity": "2468547@monolith", + "requestId": "d3d3c3c3-5c9f-4570-8d8b-cce1f5856a4f", + "historySizeBytes": "1266", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + } + } + }, + { + "eventId": "12", + "eventTime": "2026-07-22T17:46:09.043328426Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048619", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "10", + "startedEventId": "11", + "identity": "2468547@monolith", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + }, + "sdkMetadata": {}, + "meteringMetadata": {} + } + }, + { + "eventId": "13", + "eventTime": "2026-07-22T17:46:09.146751311Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED", + "taskId": "1048621", + "workflowExecutionSignaledEventAttributes": { + "signalName": "dosig", + "input": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "IjEi" + } + ] + }, + "identity": "2468547@monolith", + "requestId": "f18a65f6-8773-425d-b8f6-d1099056f758" + } + }, + { + "eventId": "14", + "eventTime": "2026-07-22T17:46:09.146757799Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048622", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "2468547@monolith-3a1c4b9f82644345a21e1663c393c0d0", + "kind": "TASK_QUEUE_KIND_STICKY", + "normalName": "tq-7049fd3e-356e-49d5-9047-2b97b454562e" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "15", + "eventTime": "2026-07-22T17:46:09.147668735Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048626", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "14", + "identity": "2468547@monolith", + "requestId": "7f9aa9f0-4bef-44a3-93f4-744ff1db8eb4", + "historySizeBytes": "1724", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + } + } + }, + { + "eventId": "16", + "eventTime": "2026-07-22T17:46:09.151221140Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048630", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "14", + "startedEventId": "15", + "identity": "2468547@monolith", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + }, + "sdkMetadata": {}, + "meteringMetadata": {} + } + }, + { + "eventId": "17", + "eventTime": "2026-07-22T17:46:09.151638714Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048633", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "2468547@monolith-3a1c4b9f82644345a21e1663c393c0d0", + "kind": "TASK_QUEUE_KIND_STICKY", + "normalName": "tq-7049fd3e-356e-49d5-9047-2b97b454562e" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "18", + "eventTime": "2026-07-22T17:46:09.151645790Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048634", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "17", + "identity": "2468547@monolith", + "requestId": "request-from-RespondWorkflowTaskCompleted", + "historySizeBytes": "1933", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + } + } + }, + { + "eventId": "19", + "eventTime": "2026-07-22T17:46:09.155070433Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048635", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "17", + "startedEventId": "18", + "identity": "2468547@monolith", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + }, + "sdkMetadata": {}, + "meteringMetadata": {} + } + }, + { + "eventId": "20", + "eventTime": "2026-07-22T17:46:09.155126964Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ACCEPTED", + "taskId": "1048636", + "workflowExecutionUpdateAcceptedEventAttributes": { + "protocolInstanceId": "dd163639-396a-440a-b950-00ac2f2ee490", + "acceptedRequestMessageId": "dd163639-396a-440a-b950-00ac2f2ee490/request", + "acceptedRequestSequencingEventId": "17", + "acceptedRequest": { + "meta": { + "updateId": "dd163639-396a-440a-b950-00ac2f2ee490", + "identity": "2468547@monolith" + }, + "input": { + "name": "doupdate", + "args": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "IjEi" + } + ] + } + } + } + } + }, + { + "eventId": "21", + "eventTime": "2026-07-22T17:46:09.155192377Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_COMPLETED", + "taskId": "1048637", + "workflowExecutionUpdateCompletedEventAttributes": { + "meta": { + "updateId": "dd163639-396a-440a-b950-00ac2f2ee490", + "identity": "2468547@monolith" + }, + "outcome": { + "success": { + "payloads": [ + { + "metadata": { + "encoding": "YmluYXJ5L251bGw=" + } + } + ] + } + }, + "acceptedEventId": "20" + } + }, + { + "eventId": "22", + "eventTime": "2026-07-22T17:46:09.801318803Z", + "eventType": "EVENT_TYPE_TIMER_FIRED", + "taskId": "1048640", + "timerFiredEventAttributes": { + "timerId": "1", + "startedEventId": "6" + } + }, + { + "eventId": "23", + "eventTime": "2026-07-22T17:46:09.801325729Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048641", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "2468547@monolith-3a1c4b9f82644345a21e1663c393c0d0", + "kind": "TASK_QUEUE_KIND_STICKY", + "normalName": "tq-7049fd3e-356e-49d5-9047-2b97b454562e" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "24", + "eventTime": "2026-07-22T17:46:09.802775220Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048645", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "23", + "identity": "2468547@monolith", + "requestId": "3d1c5327-09f4-4434-8880-f181a287c207", + "historySizeBytes": "2770", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + } + } + }, + { + "eventId": "25", + "eventTime": "2026-07-22T17:46:09.808121470Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048649", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "23", + "startedEventId": "24", + "identity": "2468547@monolith", + "workerVersion": { + "buildId": "d7dd3d4f7dae90ccfe98229c9f116c9a" + }, + "sdkMetadata": {}, + "meteringMetadata": {} + } + }, + { + "eventId": "26", + "eventTime": "2026-07-22T17:46:09.808174063Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED", + "taskId": "1048650", + "workflowExecutionCompletedEventAttributes": { + "result": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "WyJzaWctYmVmb3JlLXN5bmMiLCJzaWctYmVmb3JlLTEiLCJ0aW1lci1zeW5jIiwiYWN0LXN5bmMiLCJzaWctYmVmb3JlLTIiLCJhY3QtMSIsImFjdC0yIiwic2lnLTEtc3luYyIsInNpZy0xLTEiLCJzaWctMS0yIiwidXBkYXRlLTEtc3luYyIsInVwZGF0ZS0xLTEiLCJ1cGRhdGUtMS0yIiwidGltZXItMSIsInRpbWVyLTIiXQ==" + } + ] + }, + "workflowTaskCompletedEventId": "25" + } + } + ] +} + diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index dbd85ef20..716b497cb 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -41,6 +41,7 @@ import temporalio.converter._extstore import temporalio.worker import temporalio.worker._command_aware_visitor +import temporalio.worker._workflow_instance import temporalio.workflow from temporalio import activity, workflow from temporalio.api.common.v1 import Payload, Payloads, WorkflowExecution @@ -114,6 +115,7 @@ from temporalio.worker import ( ExecuteWorkflowInput, HandleSignalInput, + Replayer, UnsandboxedWorkflowRunner, Worker, WorkflowInstance, @@ -1246,7 +1248,99 @@ async def run(self) -> str: return "cancelled" -async def test_workflow_cancel_before_run(client: Client): +_WorkflowLogicFlag = temporalio.worker._workflow_instance._WorkflowLogicFlag +_SINGLE_BATCH_WORKFLOW_LOGIC_FLAG = ( + _WorkflowLogicFlag.PROCESS_WORKFLOW_ACTIVATION_JOBS_AS_SINGLE_BATCH +) + + +def test_single_batch_workflow_activation_jobs_default_disabled() -> None: + assert ( + _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG + not in temporalio.worker._workflow_instance._DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS + ) + + +@workflow.defn +class EnableWorkflowLogicFlagAfterReplayWorkflow: + def __init__(self) -> None: + self._ready = False + self._finish = False + + @workflow.run + async def run(self) -> str: + self._ready = True + await workflow.wait_condition(lambda: self._finish) + return "done" + + @workflow.signal + def finish(self) -> None: + self._finish = True + + @workflow.query + def ready(self) -> bool: + return self._ready + + +async def test_workflow_logic_flag_enabled_after_replay(client: Client) -> None: + task_queue = str(uuid.uuid4()) + handle = await client.start_workflow( + EnableWorkflowLogicFlagAfterReplayWorkflow.run, + id=f"workflow-{uuid.uuid4()}", + task_queue=task_queue, + ) + + async with new_worker( + client, + EnableWorkflowLogicFlagAfterReplayWorkflow, + task_queue=task_queue, + ): + + async def ready() -> bool: + return await handle.query(EnableWorkflowLogicFlagAfterReplayWorkflow.ready) + + await assert_eq_eventually(True, ready) + + await handle.signal(EnableWorkflowLogicFlagAfterReplayWorkflow.finish) + + runner = CustomWorkflowRunner() + worker = new_worker( + client, + EnableWorkflowLogicFlagAfterReplayWorkflow, + task_queue=task_queue, + workflow_runner=runner, + ) + worker._set_default_workflow_logic_flag( + _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG, enabled=True + ) + async with worker: + assert await handle.result() == "done" + + assert any(activation.is_replaying for activation, _ in runner._pairs) + assert any( + not activation.is_replaying + and _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG + in completion.successful.used_internal_flags + for activation, completion in runner._pairs + ) + + history = await handle.fetch_history() + workflow_task_flags = [ + event.workflow_task_completed_event_attributes.sdk_metadata.lang_used_flags + for event in history.events + if event.HasField("workflow_task_completed_event_attributes") + ] + assert _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG not in workflow_task_flags[0] + assert any( + _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG in flags for flags in workflow_task_flags[1:] + ) + await Replayer( + workflows=[EnableWorkflowLogicFlagAfterReplayWorkflow] + ).replay_workflow(history) + + +@pytest.mark.parametrize("single_batch", [False, True]) +async def test_workflow_cancel_before_run(client: Client, single_batch: bool): # Start the workflow _and_ send cancel before even starting the workflow task_queue = str(uuid.uuid4()) handle = await client.start_workflow( @@ -1256,10 +1350,240 @@ async def test_workflow_cancel_before_run(client: Client): ) await handle.cancel() # Start worker and wait for result - async with new_worker(client, TrapCancelWorkflow, task_queue=task_queue): + worker = new_worker(client, TrapCancelWorkflow, task_queue=task_queue) + if single_batch: + worker._set_default_workflow_logic_flag( + _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG, enabled=True + ) + async with worker: assert "cancelled" == await handle.result() +@workflow.defn +class CancelAtWaitConditionWorkflow: + def __init__(self) -> None: + self._ready = False + self._proceed = False + self._waiting_after_cancel = False + self._finish_after_cancel = False + + @workflow.run + async def run(self, timeout: bool) -> str: + self._ready = True + try: + await workflow.wait_condition( + lambda: self._proceed, timeout=1000 if timeout else None + ) + except asyncio.CancelledError: + # A caught cancellation must not be raised again merely because a + # later wait sees the already-recorded workflow cancellation. + self._waiting_after_cancel = True + await workflow.wait_condition(lambda: self._finish_after_cancel) + return "cancelled" + return "condition" + + @workflow.signal + def proceed(self) -> None: + self._proceed = True + + @workflow.signal + def finish_after_cancel(self) -> None: + self._finish_after_cancel = True + + @workflow.query + def ready(self) -> bool: + return self._ready + + @workflow.query + def waiting_after_cancel(self) -> bool: + return self._waiting_after_cancel + + +@pytest.mark.parametrize("timeout", [False, True]) +async def test_workflow_cancel_and_condition_ready_in_same_activation( + client: Client, timeout: bool +): + task_queue = str(uuid.uuid4()) + runner = CustomWorkflowRunner() + handle = await client.start_workflow( + CancelAtWaitConditionWorkflow.run, + timeout, + id=f"workflow-{uuid.uuid4()}", + task_queue=task_queue, + ) + + worker = new_worker( + client, + CancelAtWaitConditionWorkflow, + task_queue=task_queue, + workflow_runner=runner, + ) + worker._set_default_workflow_logic_flag( + _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG, enabled=True + ) + async with worker: + + async def ready() -> bool: + return await handle.query(CancelAtWaitConditionWorkflow.ready) + + await assert_eq_eventually(True, ready) + + # Keep the worker offline so the signal and cancellation are delivered in + # one activation when polling resumes. + await handle.signal(CancelAtWaitConditionWorkflow.proceed) + await handle.cancel() + + worker = new_worker( + client, + CancelAtWaitConditionWorkflow, + task_queue=task_queue, + workflow_runner=runner, + ) + worker._set_default_workflow_logic_flag( + _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG, enabled=True + ) + async with worker: + + async def waiting_after_cancel() -> bool: + return await handle.query( + CancelAtWaitConditionWorkflow.waiting_after_cancel + ) + + await assert_eq_eventually(True, waiting_after_cancel) + await handle.signal(CancelAtWaitConditionWorkflow.finish_after_cancel) + assert await handle.result() == "cancelled" + + assert any( + _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG in completion.successful.used_internal_flags + for _, completion in runner._pairs + ) + assert any( + {"signal_workflow", "cancel_workflow"}.issubset( + {job.WhichOneof("variant") for job in activation.jobs} + ) + for activation, _ in runner._pairs + ) + await Replayer(workflows=[CancelAtWaitConditionWorkflow]).replay_workflow( + await handle.fetch_history() + ) + + +@workflow.defn +class CompleteChildOnSignalWorkflow: + def __init__(self) -> None: + self._finish = False + + @workflow.run + async def run(self) -> str: + await workflow.wait_condition(lambda: self._finish) + return "child complete" + + @workflow.signal + def finish(self) -> None: + self._finish = True + + +@workflow.defn +class CancelAtChildCompletionWorkflow: + def __init__(self) -> None: + self._child_started = False + + @workflow.run + async def run(self, child_task_queue: str) -> str: + child = await workflow.start_child_workflow( + CompleteChildOnSignalWorkflow.run, + id=f"{workflow.info().workflow_id}-child", + task_queue=child_task_queue, + ) + self._child_started = True + return await child + + @workflow.query + def child_started(self) -> bool: + return self._child_started + + +async def test_workflow_cancel_and_child_completion_in_same_activation( + client: Client, +): + parent_task_queue = str(uuid.uuid4()) + child_task_queue = str(uuid.uuid4()) + workflow_id = f"workflow-{uuid.uuid4()}" + child_id = f"{workflow_id}-child" + runner = CustomWorkflowRunner() + + child_worker = new_worker( + client, + CompleteChildOnSignalWorkflow, + task_queue=child_task_queue, + ) + child_worker._set_default_workflow_logic_flag( + _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG, enabled=True + ) + async with child_worker: + handle = await client.start_workflow( + CancelAtChildCompletionWorkflow.run, + child_task_queue, + id=workflow_id, + task_queue=parent_task_queue, + ) + parent_worker = new_worker( + client, + CancelAtChildCompletionWorkflow, + task_queue=parent_task_queue, + workflow_runner=runner, + ) + parent_worker._set_default_workflow_logic_flag( + _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG, enabled=True + ) + async with parent_worker: + + async def child_started() -> bool: + return await handle.query(CancelAtChildCompletionWorkflow.child_started) + + await assert_eq_eventually(True, child_started) + + child_handle = client.get_workflow_handle(child_id) + await child_handle.signal(CompleteChildOnSignalWorkflow.finish) + assert await child_handle.result() == "child complete" + + async def child_completion_recorded() -> None: + async for event in handle.fetch_history_events(): + if ( + event.event_type + == EventType.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_COMPLETED + ): + return + raise AssertionError("Child completion is not in parent history") + + await assert_eventually(child_completion_recorded) + await handle.cancel() + + parent_worker = new_worker( + client, + CancelAtChildCompletionWorkflow, + task_queue=parent_task_queue, + workflow_runner=runner, + ) + parent_worker._set_default_workflow_logic_flag( + _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG, enabled=True + ) + async with parent_worker: + with pytest.raises(WorkflowFailureError) as err: + await handle.result() + assert isinstance(err.value.cause, CancelledError) + + assert any( + {"resolve_child_workflow_execution", "cancel_workflow"}.issubset( + {job.WhichOneof("variant") for job in activation.jobs} + ) + for activation, _ in runner._pairs + ) + await Replayer(workflows=[CancelAtChildCompletionWorkflow]).replay_workflow( + await handle.fetch_history() + ) + + @activity.defn async def wait_forever() -> NoReturn: await asyncio.Future() @@ -3945,8 +4269,16 @@ def check_condition(self) -> bool: return True -async def test_workflow_query_does_not_run_condition(client: Client): - async with new_worker(client, QueryAffectConditionWorkflow) as worker: +@pytest.mark.parametrize("single_batch", [False, True]) +async def test_workflow_query_does_not_run_condition( + client: Client, single_batch: bool +): + worker = new_worker(client, QueryAffectConditionWorkflow) + if single_batch: + worker._set_default_workflow_logic_flag( + _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG, enabled=True + ) + async with worker: handle = await client.start_workflow( QueryAffectConditionWorkflow.run, id=f"workflow-{uuid.uuid4()}", @@ -7334,8 +7666,8 @@ async def run_act(self): async def test_async_loop_ordering(client: Client, env: WorkflowEnvironment): - """This test mostly exists to generate histories for test_replayer_async_ordering. - See that test for more.""" + """This test mostly exists to generate PROCESS_WORKFLOW_ACTIVATION_JOBS_AS_SINGLE_BATCH + histories for test_replayer_async_ordering. See that test for more.""" if env.supports_time_skipping: pytest.skip("This test doesn't work right with time skipping for some reason") @@ -7347,12 +7679,16 @@ async def test_async_loop_ordering(client: Client, env: WorkflowEnvironment): ) await handle.signal(SignalsActivitiesTimersUpdatesTracingWorkflow.dosig, "before") - async with new_worker( + worker = new_worker( client, SignalsActivitiesTimersUpdatesTracingWorkflow, activities=[say_hello], task_queue=task_queue, - ): + ) + worker._set_default_workflow_logic_flag( + _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG, enabled=True + ) + async with worker: await asyncio.sleep(0.2) await handle.signal(SignalsActivitiesTimersUpdatesTracingWorkflow.dosig, "1") await handle.execute_update(