From 9ee09ba4006af76ec62f3136aec05cd947633310 Mon Sep 17 00:00:00 2001 From: Alex Mazzeo Date: Thu, 5 Nov 2026 12:19:53 -0800 Subject: [PATCH 1/2] Fix sandboxed datetime decoding with replay compatibility --- changelog/fixed/tiptoeing-clockwork.md | 1 + temporalio/_workflow_context.py | 59 +++ temporalio/converter/_payload_converter.py | 21 +- temporalio/worker/_replayer.py | 6 +- temporalio/worker/_worker.py | 2 +- temporalio/worker/_workflow.py | 6 +- temporalio/worker/_workflow_instance.py | 26 +- tests/test_workflow_context.py | 101 +++++ tests/worker/test_replayer.py | 3 +- tests/worker/test_workflow.py | 10 +- tests/worker/test_workflow_flags.py | 380 ++++++++++++++++++ ...flow_flags_datetime_unflagged_history.json | 188 +++++++++ 12 files changed, 777 insertions(+), 26 deletions(-) create mode 100644 changelog/fixed/tiptoeing-clockwork.md create mode 100644 temporalio/_workflow_context.py create mode 100644 tests/test_workflow_context.py create mode 100644 tests/worker/test_workflow_flags.py create mode 100644 tests/worker/test_workflow_flags_datetime_unflagged_history.json diff --git a/changelog/fixed/tiptoeing-clockwork.md b/changelog/fixed/tiptoeing-clockwork.md new file mode 100644 index 0000000000..9d34881cc5 --- /dev/null +++ b/changelog/fixed/tiptoeing-clockwork.md @@ -0,0 +1 @@ +Fix datetime decoding failures in sandboxed workflows, including datetime values nested in collections or dataclasses and datetime-typed signals being dropped. Existing workflow histories retain their original decoding behavior during replay. diff --git a/temporalio/_workflow_context.py b/temporalio/_workflow_context.py new file mode 100644 index 0000000000..02d07c3060 --- /dev/null +++ b/temporalio/_workflow_context.py @@ -0,0 +1,59 @@ +"""Private workflow compatibility flags and their runtime context.""" + +from __future__ import annotations + +from collections.abc import Generator +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass +from enum import IntEnum +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from temporalio.worker._workflow_instance import _WorkflowInstanceImpl + + +class _WorkflowLogicFlag(IntEnum): + """Flags recorded in workflow task metadata to preserve replay behavior.""" + + RAISE_ON_CANCELLING_COMPLETED_ACTIVITY = 1 + PROCESS_WORKFLOW_ACTIVATION_JOBS_AS_SINGLE_BATCH = 2 + DECODE_DATETIME_PROXIES = 3 + + +_DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS: frozenset[_WorkflowLogicFlag] = frozenset( + { + _WorkflowLogicFlag.PROCESS_WORKFLOW_ACTIVATION_JOBS_AS_SINGLE_BATCH, + _WorkflowLogicFlag.DECODE_DATETIME_PROXIES, + } +) + + +@dataclass(frozen=True) +class _WorkflowFlagContext: + _runtime: _WorkflowInstanceImpl + + def is_flag_enabled(self, flag: _WorkflowLogicFlag) -> bool: + """Consult live runtime state, not the state at context creation.""" + return self._runtime._workflow_logic_flag_enabled(flag) + + +_current_workflow_flags: ContextVar[_WorkflowFlagContext | None] = ContextVar( + "temporal_workflow_flags", default=None +) + + +@contextmanager +def _workflow_flag_scope( # type:ignore[reportUnusedFunction] + *, + runtime: _WorkflowInstanceImpl, +) -> Generator[None]: + """Expose live flag decisions without inheriting another workflow's state.""" + context = _WorkflowFlagContext( + _runtime=runtime, + ) + token = _current_workflow_flags.set(context) + try: + yield + finally: + _current_workflow_flags.reset(token) diff --git a/temporalio/converter/_payload_converter.py b/temporalio/converter/_payload_converter.py index 8b428c2290..81b1db34f3 100644 --- a/temporalio/converter/_payload_converter.py +++ b/temporalio/converter/_payload_converter.py @@ -43,6 +43,10 @@ # Python's datetime.fromisoformat doesn't support certain formats pre-3.11 from dateutil import parser # type: ignore +from temporalio._workflow_context import ( + _current_workflow_flags, + _WorkflowLogicFlag, +) from temporalio.converter._serialization_context import ( SerializationContext, WithSerializationContext, @@ -800,6 +804,21 @@ def from_payload( raise RuntimeError("Failed parsing") from err +def _is_datetime_hint(hint: type) -> bool: + if hint is datetime: + return True + context = _current_workflow_flags.get() + if context is None or not context.is_flag_enabled( + _WorkflowLogicFlag.DECODE_DATETIME_PROXIES + ): + return False + + # Importing the worker package during converter initialization creates a cycle. + from temporalio.worker.workflow_sandbox._restrictions import RestrictionContext + + return RestrictionContext.unwrap_if_proxied(hint) is datetime + + def _get_iso_datetime_parser() -> Callable[[str], datetime]: """Isolates system version check and returns relevant datetime passer @@ -846,7 +865,7 @@ def value_to_type( # Any or primitives if hint is Any: return value - elif hint is datetime: + elif _is_datetime_hint(hint): if isinstance(value, str): try: return _get_iso_datetime_parser()(value) diff --git a/temporalio/worker/_replayer.py b/temporalio/worker/_replayer.py index 62a27b29cf..1542b3e0f2 100644 --- a/temporalio/worker/_replayer.py +++ b/temporalio/worker/_replayer.py @@ -20,15 +20,17 @@ import temporalio.worker import temporalio.workflow +from .._workflow_context import ( + _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS, + _WorkflowLogicFlag, +) from ..common import HeaderCodecBehavior from ._interceptor import Interceptor from ._worker import load_default_build_id from ._workflow import _WorkflowWorker from ._workflow_instance import ( - _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS, UnsandboxedWorkflowRunner, WorkflowRunner, - _WorkflowLogicFlag, ) from .workflow_sandbox import SandboxedWorkflowRunner diff --git a/temporalio/worker/_worker.py b/temporalio/worker/_worker.py index e06bb99a4b..cf8125d819 100644 --- a/temporalio/worker/_worker.py +++ b/temporalio/worker/_worker.py @@ -30,6 +30,7 @@ WorkerDeploymentVersion, ) +from .._workflow_context import _WorkflowLogicFlag from ._activity import SharedStateManager, _ActivityWorker from ._interceptor import Interceptor from ._nexus import _NexusWorker @@ -43,7 +44,6 @@ PatchActivationInput, UnsandboxedWorkflowRunner, WorkflowRunner, - _WorkflowLogicFlag, ) from .workflow_sandbox import SandboxedWorkflowRunner diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py index 64c8a323b3..1795853cc6 100644 --- a/temporalio/worker/_workflow.py +++ b/temporalio/worker/_workflow.py @@ -30,6 +30,10 @@ from temporalio.converter import StorageDriverStoreContext, StorageDriverWorkflowInfo from temporalio.worker.workflow_sandbox._runner import SandboxedWorkflowRunner +from .._workflow_context import ( + _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS, + _WorkflowLogicFlag, +) from . import _command_aware_visitor from ._debugger import ( _install_workflow_breakpoint_hook, @@ -41,13 +45,11 @@ WorkflowInterceptorClassInput, ) from ._workflow_instance import ( - _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS, PatchActivationInput, WorkflowInstance, WorkflowInstanceDetails, WorkflowRunner, _WorkflowExternFunctions, - _WorkflowLogicFlag, ) logger = logging.getLogger(__name__) diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 7bdf7faf75..82f5d4e66e 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -29,7 +29,6 @@ from contextlib import contextmanager from dataclasses import dataclass, field from datetime import timedelta -from enum import IntEnum from typing import ( Any, Generic, @@ -67,6 +66,11 @@ ) from temporalio.service import __version__ +from .._workflow_context import ( + _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS, + _workflow_flag_scope, + _WorkflowLogicFlag, +) from ..api.failure.v1.message_pb2 import Failure from ..workflow._event_groups import ( _capture_event_group_markers, @@ -477,6 +481,12 @@ def get_thread_id(self) -> int | None: def activate( self, act: temporalio.bridge.proto.workflow_activation.WorkflowActivation + ) -> temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion: + with _workflow_flag_scope(runtime=self): + return self._activate(act) + + def _activate( + self, act: temporalio.bridge.proto.workflow_activation.WorkflowActivation ) -> temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion: # Reset current completion, time, and whether replaying self._current_completion = ( @@ -504,6 +514,8 @@ def activate( self._single_batch_activation = self._workflow_logic_flag_enabled( _WorkflowLogicFlag.PROCESS_WORKFLOW_ACTIVATION_JOBS_AS_SINGLE_BATCH ) + # Record this before queries can arrive, even when no input needs decoding. + self._workflow_logic_flag_enabled(_WorkflowLogicFlag.DECODE_DATETIME_PROXIES) activation_err: Exception | None = None try: # Split into job sets with patches, then signals + updates, then @@ -4251,15 +4263,3 @@ def _make_unfinished_signal_handler_message( [{"name": name, "count": count} for name, count in names.most_common()] ) ) - - -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 - - -_DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS: frozenset[_WorkflowLogicFlag] = frozenset( - {_WorkflowLogicFlag.PROCESS_WORKFLOW_ACTIVATION_JOBS_AS_SINGLE_BATCH} -) diff --git a/tests/test_workflow_context.py b/tests/test_workflow_context.py new file mode 100644 index 0000000000..35cd407233 --- /dev/null +++ b/tests/test_workflow_context.py @@ -0,0 +1,101 @@ +from datetime import datetime +from typing import Any +from unittest.mock import create_autospec + +import pytest + +import temporalio.converter +from temporalio._workflow_context import ( + _current_workflow_flags, + _workflow_flag_scope, + _WorkflowLogicFlag, +) +from temporalio.worker._workflow_instance import _WorkflowInstanceImpl +from temporalio.worker.workflow_sandbox._importer import Importer +from temporalio.worker.workflow_sandbox._restrictions import ( + RestrictionContext, + SandboxRestrictions, +) + + +class FlagRuntime: + def __init__(self, enabled: bool) -> None: + self.enabled = enabled + self.checked: list[_WorkflowLogicFlag] = [] + instance = create_autospec(_WorkflowInstanceImpl, instance=True) + instance._workflow_logic_flag_enabled.side_effect = ( + self._workflow_logic_flag_enabled + ) + self.instance: _WorkflowInstanceImpl = instance + + def _workflow_logic_flag_enabled(self, flag: _WorkflowLogicFlag) -> bool: + self.checked.append(flag) + return self.enabled + + +def datetime_hint() -> type: + with Importer(SandboxRestrictions.default, RestrictionContext()).applied(): + from datetime import datetime as sandboxed_datetime + + return sandboxed_datetime + + +def test_workflow_flag_scope_restores_context_after_exception() -> None: + outer, inner = FlagRuntime(True), FlagRuntime(False) + assert _current_workflow_flags.get() is None + with _workflow_flag_scope(runtime=outer.instance): + context = _current_workflow_flags.get() + assert context is not None + assert context.is_flag_enabled(_WorkflowLogicFlag.DECODE_DATETIME_PROXIES) + with ( + pytest.raises(ValueError), + _workflow_flag_scope(runtime=inner.instance), + ): + nested = _current_workflow_flags.get() + assert nested is not None + assert not nested.is_flag_enabled( + _WorkflowLogicFlag.DECODE_DATETIME_PROXIES + ) + raise ValueError("test cleanup") + assert _current_workflow_flags.get() is context + assert _current_workflow_flags.get() is None + + +def test_datetime_proxy_conversion_policy() -> None: + hint = datetime_hint() + assert hint is not datetime + value = "2021-01-02T00:00:00+00:00" + with pytest.raises(TypeError): + temporalio.converter.value_to_type(hint, value) + + disabled = FlagRuntime(False) + with _workflow_flag_scope(runtime=disabled.instance): + with pytest.raises(TypeError): + temporalio.converter.value_to_type(hint, value) + assert disabled.checked == [_WorkflowLogicFlag.DECODE_DATETIME_PROXIES] + + enabled = FlagRuntime(True) + with _workflow_flag_scope(runtime=enabled.instance): + assert temporalio.converter.value_to_type( + hint, value + ) == datetime.fromisoformat(value) + assert enabled.checked == [_WorkflowLogicFlag.DECODE_DATETIME_PROXIES] + + +def test_datetime_proxy_custom_converter_precedence() -> None: + expected_hint = datetime_hint() + + class CustomConverter(temporalio.converter.JSONTypeConverter): + def to_typed_value(self, hint: type, value: Any) -> Any: + assert hint is expected_hint + return "custom" + + runtime = FlagRuntime(True) + with _workflow_flag_scope(runtime=runtime.instance): + assert ( + temporalio.converter.value_to_type( + expected_hint, "input", [CustomConverter()] + ) + == "custom" + ) + assert not runtime.checked diff --git a/tests/worker/test_replayer.py b/tests/worker/test_replayer.py index b0274ca46b..8f4b1fd507 100644 --- a/tests/worker/test_replayer.py +++ b/tests/worker/test_replayer.py @@ -10,8 +10,8 @@ import pytest -import temporalio.worker._workflow_instance from temporalio import activity, workflow +from temporalio._workflow_context import _WorkflowLogicFlag from temporalio.client import Client, WorkflowFailureError, WorkflowHistory from temporalio.exceptions import ApplicationError from temporalio.testing import WorkflowEnvironment @@ -82,7 +82,6 @@ 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 ) diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index 5b5280bfba..842093f776 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -45,6 +45,10 @@ import temporalio.worker._workflow_instance import temporalio.workflow from temporalio import activity, workflow +from temporalio._workflow_context import ( + _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS, + _WorkflowLogicFlag, +) from temporalio.api.common.v1 import Payload, Payloads, WorkflowExecution from temporalio.api.enums.v1 import EventType from temporalio.api.failure.v1 import Failure @@ -1474,17 +1478,13 @@ async def run(self) -> str: return "cancelled" -_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_enabled() -> None: - assert ( - _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG - in temporalio.worker._workflow_instance._DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS - ) + assert _SINGLE_BATCH_WORKFLOW_LOGIC_FLAG in _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS @workflow.defn diff --git a/tests/worker/test_workflow_flags.py b/tests/worker/test_workflow_flags.py new file mode 100644 index 0000000000..e4b6b0266f --- /dev/null +++ b/tests/worker/test_workflow_flags.py @@ -0,0 +1,380 @@ +import uuid +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +import pytest + +import temporalio.client +import temporalio.converter +import temporalio.exceptions +import temporalio.worker +import temporalio.worker.workflow_sandbox +import temporalio.workflow +from temporalio._workflow_context import _WorkflowLogicFlag +from tests.helpers import assert_eq_eventually + + +@temporalio.workflow.defn +class DatetimeArgumentWorkflow: + @temporalio.workflow.run + async def run(self, when: datetime) -> str: + return when.isoformat() + + +@dataclass +class DatetimeInput: + when: datetime + + +@temporalio.workflow.defn +class DatetimeDataclassWorkflow: + @temporalio.workflow.run + async def run(self, value: DatetimeInput) -> str: + return value.when.isoformat() + + +@temporalio.workflow.defn +class DatetimeListWorkflow: + @temporalio.workflow.run + async def run(self, values: list[datetime]) -> str: + return values[0].isoformat() + + +@pytest.mark.parametrize("runner", ["sandboxed", "passthrough", "unsandboxed"]) +@pytest.mark.parametrize("datetime_decoding_enabled", [False, True]) +@pytest.mark.parametrize("shape", ["bare", "list", "dataclass"]) +async def test_datetime_arguments_respect_workflow_flag( + client: temporalio.client.Client, + runner: str, + datetime_decoding_enabled: bool, + shape: str, +): + workflow_runner: temporalio.worker.WorkflowRunner + if runner == "unsandboxed": + workflow_runner = temporalio.worker.UnsandboxedWorkflowRunner() + else: + restrictions = temporalio.worker.workflow_sandbox.SandboxRestrictions.default + if runner == "passthrough": + restrictions = restrictions.with_passthrough_modules(__name__) + workflow_runner = temporalio.worker.workflow_sandbox.SandboxedWorkflowRunner( + restrictions + ) + + workflow_type = { + "bare": DatetimeArgumentWorkflow, + "list": DatetimeListWorkflow, + "dataclass": DatetimeDataclassWorkflow, + }[shape] + when = datetime(2021, 1, 2, tzinfo=timezone.utc) + argument = {"bare": when, "list": [when], "dataclass": DatetimeInput(when)}[shape] + worker = temporalio.worker.Worker( + client, + task_queue=str(uuid.uuid4()), + workflows=[workflow_type], + workflow_runner=workflow_runner, + # Fail decoding errors immediately instead of retrying workflow tasks forever. + workflow_failure_exception_types=[Exception], + ) + if not datetime_decoding_enabled: + worker._set_default_workflow_logic_flag( + _WorkflowLogicFlag.DECODE_DATETIME_PROXIES, enabled=False + ) + async with worker: + handle = await client.start_workflow( + workflow_type.__name__, + argument, + id=str(uuid.uuid4()), + task_queue=worker.task_queue, + ) + if runner == "sandboxed" and not datetime_decoding_enabled: + with pytest.raises(temporalio.client.WorkflowFailureError) as err: + await handle.result() + failure = err.value.cause + while ( + isinstance(failure, temporalio.exceptions.FailureError) + and failure.cause + ): + failure = failure.cause + assert failure + assert "Unserializable type during conversion" in str(failure) + else: + assert await handle.result() == when.isoformat() + + history = await handle.fetch_history() + used_flags = { + flag + for event in history.events + if event.HasField("workflow_task_completed_event_attributes") + for flag in event.workflow_task_completed_event_attributes.sdk_metadata.lang_used_flags + } + assert (_WorkflowLogicFlag.DECODE_DATETIME_PROXIES in used_flags) == ( + datetime_decoding_enabled + ) + await temporalio.worker.Replayer( + workflows=[workflow_type], + workflow_runner=workflow_runner, + workflow_failure_exception_types=[Exception], + ).replay_workflow(history) + + +@temporalio.workflow.defn +class DatetimeSignalWorkflow: + def __init__(self) -> None: + self.when = "" + + @temporalio.workflow.run + async def run(self) -> str: + try: + await temporalio.workflow.wait_condition( + lambda: bool(self.when), timeout=timedelta(milliseconds=200) + ) + except TimeoutError: + return "dropped" + await temporalio.workflow.sleep(0.01) + # An extra command makes processing a formerly dropped signal detectable on replay. + await temporalio.workflow.sleep(0.01) + return self.when + + @temporalio.workflow.signal + def set_when(self, when: datetime) -> None: + self.when = when.isoformat() + + +async def test_datetime_signal_records_workflow_flag( + client: temporalio.client.Client, +) -> None: + worker = temporalio.worker.Worker( + client, + task_queue=str(uuid.uuid4()), + workflows=[DatetimeSignalWorkflow], + ) + when = datetime(2021, 1, 2, tzinfo=timezone.utc) + handle = await client.start_workflow( + DatetimeSignalWorkflow.run, + id=str(uuid.uuid4()), + task_queue=worker.task_queue, + ) + await handle.signal(DatetimeSignalWorkflow.set_when, when) + async with worker: + assert await handle.result() == when.isoformat() + + history = await handle.fetch_history() + assert any( + _WorkflowLogicFlag.DECODE_DATETIME_PROXIES + in event.workflow_task_completed_event_attributes.sdk_metadata.lang_used_flags + for event in history.events + if event.HasField("workflow_task_completed_event_attributes") + ) + await temporalio.worker.Replayer( + workflows=[DatetimeSignalWorkflow] + ).replay_workflow(history) + + +def _load_unflagged_datetime_history() -> temporalio.client.WorkflowHistory: + return temporalio.client.WorkflowHistory.from_json( + "datetime-signal-legacy", + Path(__file__) + .with_name("test_workflow_flags_datetime_unflagged_history.json") + .read_text(), + ) + + +async def test_replay_preserves_dropped_datetime_signal_without_flag() -> None: + history = _load_unflagged_datetime_history() + completed_tasks = [ + event.workflow_task_completed_event_attributes + for event in history.events + if event.HasField("workflow_task_completed_event_attributes") + ] + assert completed_tasks + assert all( + _WorkflowLogicFlag.DECODE_DATETIME_PROXIES + not in completed.sdk_metadata.lang_used_flags + for completed in completed_tasks + ) + result = await temporalio.worker.Replayer( + workflows=[DatetimeSignalWorkflow] + ).replay_workflow(history, raise_on_replay_failure=False) + assert result.replay_failure is None + completion = result.history.events[-1].workflow_execution_completed_event_attributes + assert temporalio.converter.DefaultPayloadConverter().from_payloads( + completion.result.payloads + ) == ["dropped"] + + +@temporalio.workflow.defn(name="DatetimeSignalWorkflow") +class UntypedDatetimeSignalWorkflow(DatetimeSignalWorkflow): + @temporalio.workflow.run + async def run(self) -> str: + return await super().run() + + @temporalio.workflow.signal + def set_when(self, when: Any) -> None: + # Without a datetime hint, the recorded JSON string reaches the handler. + self.when = when + + +async def test_replay_rejects_handling_previously_dropped_datetime_signal() -> None: + history = _load_unflagged_datetime_history() + original_history = history.to_json() + result = await temporalio.worker.Replayer( + workflows=[UntypedDatetimeSignalWorkflow] + ).replay_workflow(history, raise_on_replay_failure=False) + assert isinstance(result.replay_failure, temporalio.workflow.NondeterminismError) + assert history.to_json() == original_history + + +@temporalio.workflow.defn +class DatetimeMemoWorkflow: + def __init__(self) -> None: + self.ready = False + self.finished = False + + @temporalio.workflow.run + async def run(self, read_after_signal: bool) -> str: + self.ready = True + if read_after_signal: + await temporalio.workflow.wait_condition(lambda: self.finished) + return temporalio.workflow.memo_value("when", type_hint=datetime).isoformat() + + @temporalio.workflow.query + def is_ready(self) -> bool: + return self.ready + + @temporalio.workflow.signal + def finish(self) -> None: + self.finished = True + + +@pytest.mark.parametrize("read_after_signal", [False, True]) +@pytest.mark.parametrize("datetime_decoding_enabled", [False, True]) +async def test_workflow_flag_applies_to_typed_memo( + client: temporalio.client.Client, + read_after_signal: bool, + datetime_decoding_enabled: bool, +) -> None: + worker = temporalio.worker.Worker( + client, + task_queue=str(uuid.uuid4()), + workflows=[DatetimeMemoWorkflow], + workflow_failure_exception_types=[Exception], + ) + if not datetime_decoding_enabled: + worker._set_default_workflow_logic_flag( + _WorkflowLogicFlag.DECODE_DATETIME_PROXIES, enabled=False + ) + when = datetime(2021, 1, 2, tzinfo=timezone.utc) + async with worker: + handle = await client.start_workflow( + DatetimeMemoWorkflow.run, + read_after_signal, + id=str(uuid.uuid4()), + task_queue=worker.task_queue, + memo={"when": when}, + ) + if read_after_signal: + + async def ready() -> bool: + return await handle.query(DatetimeMemoWorkflow.is_ready) + + await assert_eq_eventually(True, ready) + await handle.signal(DatetimeMemoWorkflow.finish) + if datetime_decoding_enabled: + assert await handle.result() == when.isoformat() + else: + with pytest.raises(temporalio.client.WorkflowFailureError): + await handle.result() + history = await handle.fetch_history() + if read_after_signal: + 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 (_WorkflowLogicFlag.DECODE_DATETIME_PROXIES in task_flags[0]) == ( + datetime_decoding_enabled + ) + assert len(task_flags) >= 2 + await temporalio.worker.Replayer( + workflows=[DatetimeMemoWorkflow], + workflow_failure_exception_types=[Exception], + ).replay_workflow(history) + + +@temporalio.workflow.defn +class DatetimeQueryWorkflow: + def __init__(self) -> None: + self.when = "" + self.finished = False + + @temporalio.workflow.run + async def run(self) -> None: + await temporalio.workflow.wait_condition(lambda: self.finished) + + @temporalio.workflow.query + def echo(self, when: datetime) -> str: + return when.isoformat() + + @temporalio.workflow.query + def state(self) -> str: + return self.when + + @temporalio.workflow.signal + def set_when(self, when: datetime) -> None: + self.when = when.isoformat() + + @temporalio.workflow.signal + def finish(self) -> None: + self.finished = True + + +async def test_new_workflow_records_default_flags_before_first_datetime_query( + client: temporalio.client.Client, +) -> None: + worker = temporalio.worker.Worker( + client, + task_queue=str(uuid.uuid4()), + workflows=[DatetimeQueryWorkflow], + ) + when = datetime(2021, 1, 2, tzinfo=timezone.utc) + async with worker: + handle = await client.start_workflow( + DatetimeQueryWorkflow.run, + id=str(uuid.uuid4()), + task_queue=worker.task_queue, + ) + # This workflow has no inputs, and this query does not decode a datetime. + assert await handle.query(DatetimeQueryWorkflow.state) == "" + history = await handle.fetch_history() + completed_tasks = [ + event.workflow_task_completed_event_attributes + for event in history.events + if event.HasField("workflow_task_completed_event_attributes") + ] + assert completed_tasks + assert _WorkflowLogicFlag.DECODE_DATETIME_PROXIES in ( + completed_tasks[0].sdk_metadata.lang_used_flags + ) + # The first datetime query can rely on the flag recorded by the initial task. + assert await handle.query(DatetimeQueryWorkflow.echo, when) == when.isoformat() + await handle.signal(DatetimeQueryWorkflow.set_when, when) + + async def state() -> str: + return await handle.query(DatetimeQueryWorkflow.state) + + await assert_eq_eventually(when.isoformat(), state) + assert await handle.query(DatetimeQueryWorkflow.echo, when) == when.isoformat() + await handle.signal(DatetimeQueryWorkflow.finish) + await handle.result() + history = await handle.fetch_history() + assert any( + _WorkflowLogicFlag.DECODE_DATETIME_PROXIES + in event.workflow_task_completed_event_attributes.sdk_metadata.lang_used_flags + for event in history.events + if event.HasField("workflow_task_completed_event_attributes") + ) + await temporalio.worker.Replayer(workflows=[DatetimeQueryWorkflow]).replay_workflow( + history + ) diff --git a/tests/worker/test_workflow_flags_datetime_unflagged_history.json b/tests/worker/test_workflow_flags_datetime_unflagged_history.json new file mode 100644 index 0000000000..7d9c216dd8 --- /dev/null +++ b/tests/worker/test_workflow_flags_datetime_unflagged_history.json @@ -0,0 +1,188 @@ +{ + "events": [ + { + "eventId": "1", + "eventTime": "2026-11-05T17:50:50.926524Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_STARTED", + "taskId": "1048587", + "workflowExecutionStartedEventAttributes": { + "workflowType": { + "name": "DatetimeSignalWorkflow" + }, + "taskQueue": { + "name": "datetime-signal-legacy", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "workflowTaskTimeout": "10s", + "originalExecutionRunId": "01a1acd5-bc2e-77fa-8e70-9afb25bc7ed5", + "identity": "datetime-replay-recorder", + "firstExecutionRunId": "01a1acd5-bc2e-77fa-8e70-9afb25bc7ed5", + "attempt": 1, + "firstWorkflowTaskBackoff": "0s", + "workflowId": "datetime-signal-legacy", + "priority": {} + } + }, + { + "eventId": "2", + "eventTime": "2026-11-05T17:50:50.926563Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048588", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "datetime-signal-legacy", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "3", + "eventTime": "2026-11-05T17:50:50.928961Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED", + "taskId": "1048593", + "workflowExecutionSignaledEventAttributes": { + "signalName": "set_when", + "input": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "IjIwMjEtMDEtMDJUMDA6MDA6MDArMDA6MDAi" + } + ] + }, + "identity": "datetime-replay-recorder", + "requestId": "e69636e5-ad68-4b4d-b697-d9955e9dee9a" + } + }, + { + "eventId": "4", + "eventTime": "2026-11-05T17:50:50.932379Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048595", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "2", + "identity": "datetime-replay-recorder", + "requestId": "d3c502dd-b7f6-4ef4-abcd-5d53e3c39d44", + "historySizeBytes": "453", + "workerVersion": { + "buildId": "337e31418f35d259824f44a34bf11cb0" + } + } + }, + { + "eventId": "5", + "eventTime": "2026-11-05T17:50:50.970580Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048599", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "2", + "startedEventId": "4", + "identity": "datetime-replay-recorder", + "workerVersion": { + "buildId": "337e31418f35d259824f44a34bf11cb0" + }, + "sdkMetadata": { + "coreUsedFlags": [ + 1, + 2, + 4, + 3 + ], + "sdkName": "temporal-python", + "sdkVersion": "1.35.0" + }, + "meteringMetadata": {} + } + }, + { + "eventId": "6", + "eventTime": "2026-11-05T17:50:50.970709Z", + "eventType": "EVENT_TYPE_TIMER_STARTED", + "taskId": "1048600", + "timerStartedEventAttributes": { + "timerId": "1", + "startToFireTimeout": "0.200s", + "workflowTaskCompletedEventId": "5" + } + }, + { + "eventId": "7", + "eventTime": "2026-11-05T17:50:51.929385Z", + "eventType": "EVENT_TYPE_TIMER_FIRED", + "taskId": "1048604", + "timerFiredEventAttributes": { + "timerId": "1", + "startedEventId": "6" + } + }, + { + "eventId": "8", + "eventTime": "2026-11-05T17:50:51.929484Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048605", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "datetime-replay-sticky", + "kind": "TASK_QUEUE_KIND_STICKY", + "normalName": "datetime-signal-legacy" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "9", + "eventTime": "2026-11-05T17:50:51.930447Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048609", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "8", + "identity": "datetime-replay-recorder", + "requestId": "e74422d1-bf7f-4ca3-84b6-93b3c36d2ace", + "historySizeBytes": "918", + "workerVersion": { + "buildId": "337e31418f35d259824f44a34bf11cb0" + } + } + }, + { + "eventId": "10", + "eventTime": "2026-11-05T17:50:51.933497Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048613", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "8", + "startedEventId": "9", + "identity": "datetime-replay-recorder", + "workerVersion": { + "buildId": "337e31418f35d259824f44a34bf11cb0" + }, + "sdkMetadata": {}, + "meteringMetadata": {} + } + }, + { + "eventId": "11", + "eventTime": "2026-11-05T17:50:51.933644Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED", + "taskId": "1048614", + "workflowExecutionCompletedEventAttributes": { + "result": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "ImRyb3BwZWQi" + } + ] + }, + "workflowTaskCompletedEventId": "10" + } + } + ] +} From 65f45501313267e7782a39f481100c1a702449fa Mon Sep 17 00:00:00 2001 From: Alex Mazzeo Date: Thu, 5 Nov 2026 15:37:03 -0800 Subject: [PATCH 2/2] Fix replay timeout handling on Python 3.10 --- tests/worker/test_workflow_flags.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/worker/test_workflow_flags.py b/tests/worker/test_workflow_flags.py index e4b6b0266f..18c98984a4 100644 --- a/tests/worker/test_workflow_flags.py +++ b/tests/worker/test_workflow_flags.py @@ -1,3 +1,4 @@ +import asyncio import uuid from dataclasses import dataclass from datetime import datetime, timedelta, timezone @@ -130,7 +131,7 @@ async def run(self) -> str: await temporalio.workflow.wait_condition( lambda: bool(self.when), timeout=timedelta(milliseconds=200) ) - except TimeoutError: + except asyncio.TimeoutError: return "dropped" await temporalio.workflow.sleep(0.01) # An extra command makes processing a formerly dropped signal detectable on replay.