diff --git a/cadence/_internal/workflow/context.py b/cadence/_internal/workflow/context.py index 307ac23..1befb9b 100644 --- a/cadence/_internal/workflow/context.py +++ b/cadence/_internal/workflow/context.py @@ -3,10 +3,12 @@ from contextlib import contextmanager from asyncio import get_running_loop from datetime import datetime, timedelta +from json import JSONDecoder from math import ceil from typing import Iterator, Optional, Any, Unpack, Type, cast, Callable from cadence._internal.workflow.deterministic_event_loop import DeterministicEventLoop +from cadence._internal.workflow.deterministic_event_loop import FatalDecisionError from cadence._internal.workflow.memo import memo_to_proto from cadence._internal.workflow.retry_policy import retry_policy_to_proto from cadence._internal.workflow.statemachine.decision_manager import DecisionManager @@ -28,7 +30,7 @@ StartTimerDecisionAttributes, ) from cadence.api.v1.tasklist_pb2 import TaskList, TaskListKind -from cadence.data_converter import DataConverter +from cadence.data_converter import DataConverter, DefaultDataConverter from cadence.workflow import ( ActivityOptions, ChildWorkflowFuture, @@ -37,6 +39,8 @@ WorkflowCancellationInfo, WorkflowContext, WorkflowInfo, + VersioningOption, + DEFAULT_VERSION, ) from cadence.api.v1.history_pb2 import WorkflowExecutionCancelRequestedEventAttributes @@ -57,6 +61,9 @@ def __init__( self._replay_current_time: Optional[datetime] = None self._decision_manager = decision_manager self._cancellation_info: WorkflowCancellationInfo | None = None + # None represents a provisional DEFAULT_VERSION returned while replaying + # history before a Version marker becomes available. + self._versions: dict[str, int | None] = {} def info(self) -> WorkflowInfo: return self._info @@ -333,6 +340,174 @@ def mutable_side_effect( self.data_converter().from_data(result_payload, [result_type])[0], ) + def get_version( + self, + change_id: str, + min_supported: int, + max_supported: int, + *options: VersioningOption, + ) -> int: + self._validate_version_arguments(change_id, min_supported, max_supported) + + if change_id in self._versions: + cached = self._versions[change_id] + if cached is None: + details = self._decision_manager.version_marker_details(change_id) + if details is not None: + recorded = self._decode_recorded_version(change_id, details) + self._validate_recorded_version( + change_id, recorded, min_supported, max_supported, "recorded" + ) + self._versions[change_id] = recorded + if self._decision_manager.has_pending_version_marker(change_id): + self._decision_manager.record_version_marker(change_id, details) + return recorded + self._validate_recorded_version( + change_id, + DEFAULT_VERSION, + min_supported, + max_supported, + "cached", + ) + return DEFAULT_VERSION + self._validate_recorded_version( + change_id, cached, min_supported, max_supported, "cached" + ) + return cached + + details = self._decision_manager.version_marker_details(change_id) + if details is not None: + recorded = self._decode_recorded_version(change_id, details) + self._validate_recorded_version( + change_id, recorded, min_supported, max_supported, "recorded" + ) + self._versions[change_id] = recorded + # Recreate the state machine so the preloaded marker is completed + # when its output event is replayed. + if self._decision_manager.has_pending_version_marker(change_id): + self._decision_manager.record_version_marker(change_id, details) + return recorded + + # A version introduced while replaying old history must preserve the old + # code path. There is no marker to validate or state machine to create. + if self.is_replay_mode(): + self._validate_recorded_version( + change_id, + DEFAULT_VERSION, + min_supported, + max_supported, + "markerless replay", + ) + self._versions[change_id] = None + return DEFAULT_VERSION + + selected = self._select_version(min_supported, max_supported, *options) + self._validate_selected_version( + change_id, selected, min_supported, max_supported + ) + self._versions[change_id] = selected + + # DEFAULT_VERSION is intentionally never written to history. + if selected == DEFAULT_VERSION: + return selected + + self._decision_manager.record_version_marker( + change_id, self.data_converter().to_data([selected]) + ) + return selected + + @staticmethod + def _validate_version_arguments( + change_id: str, min_supported: int, max_supported: int + ) -> None: + if not isinstance(change_id, str) or not change_id: + raise ValueError("change_id must be a non-empty str") + for name, value in ( + ("min_supported", min_supported), + ("max_supported", max_supported), + ): + if isinstance(value, bool) or not isinstance(value, int): + raise ValueError(f"{name} must be an int") + if min_supported > max_supported: + raise ValueError("min_supported must not be greater than max_supported") + + @staticmethod + def _select_version( + min_supported: int, max_supported: int, *options: VersioningOption + ) -> int: + custom_version: int | None = None + use_min_version = False + for option in options: + if not isinstance(option, VersioningOption): + raise ValueError("get_version options must be VersioningOption values") + if option._kind == "version": + custom_version = option._version + elif option._kind == "min": + use_min_version = True + else: + raise ValueError("invalid get_version option") + if custom_version is not None: + return custom_version + if use_min_version: + return min_supported + return max_supported + + @staticmethod + def _validate_selected_version( + change_id: str, version: int, min_supported: int, max_supported: int + ) -> None: + if version < min_supported or version > max_supported: + raise ValueError( + f"selected version {version} for change_id {change_id!r} is outside " + f"the supported range [{min_supported}, {max_supported}]" + ) + + @staticmethod + def _validate_recorded_version( + change_id: str, + version: int, + min_supported: int, + max_supported: int, + source: str, + ) -> None: + if version < min_supported or version > max_supported: + raise FatalDecisionError( + f"{source} version {version} for change_id {change_id!r} is outside " + f"the supported range [{min_supported}, {max_supported}]" + ) + + def _decode_recorded_version(self, change_id: str, details: Payload) -> int: + if type(self.data_converter()) is DefaultDataConverter: + self._validate_default_version_details(change_id, details) + try: + version = self.data_converter().from_data(details, [int])[0] + except Exception as exc: + raise FatalDecisionError( + f"Unable to decode Version marker for change_id {change_id!r}" + ) from exc + if isinstance(version, bool) or not isinstance(version, int): + raise FatalDecisionError( + f"Version marker for change_id {change_id!r} did not contain an int" + ) + return cast(int, version) + + @staticmethod + def _validate_default_version_details(change_id: str, details: Payload) -> None: + if not details.data: + raise FatalDecisionError( + f"Version marker for change_id {change_id!r} had empty details" + ) + try: + _, end = JSONDecoder(strict=False).raw_decode(details.data.decode()) + except (UnicodeDecodeError, ValueError) as exc: + raise FatalDecisionError( + f"Version marker for change_id {change_id!r} had invalid details" + ) from exc + if end != len(details.data): + raise FatalDecisionError( + f"Version marker for change_id {change_id!r} had invalid details" + ) + def set_replay_current_time(self, current_time: datetime) -> None: """Set the current replay timestamp.""" self._replay_current_time = current_time diff --git a/cadence/_internal/workflow/statemachine/decision_manager.py b/cadence/_internal/workflow/statemachine/decision_manager.py index 7f63cdf..eebead0 100644 --- a/cadence/_internal/workflow/statemachine/decision_manager.py +++ b/cadence/_internal/workflow/statemachine/decision_manager.py @@ -32,7 +32,9 @@ ) from cadence._internal.workflow.statemachine.marker_state_machine import ( MUTABLE_SIDE_EFFECT_MARKER_NAME, + VERSION_MARKER_NAME, encode_marker_header, + has_marker_header, marker_context_id, marker_decision_id, KNOWN_MARKER_NAMES, @@ -43,6 +45,7 @@ ) from cadence._internal.workflow.statemachine.nondeterminism import DeterminismTracker from cadence._internal.workflow.statemachine.cancellation import is_immediate_cancel +from cadence._internal.workflow.deterministic_event_loop import FatalDecisionError from cadence._internal.workflow.statemachine.signal_external_workflow_state_machine import ( signal_external_events, SignalExternalWorkflowStateMachine, @@ -122,6 +125,8 @@ def __init__(self, event_loop: asyncio.AbstractEventLoop): self._determinism_tracker = DeterminismTracker() self._replaying = False self._recorded_marker_details: Dict[DecisionId, Payload] = {} + self._version_marker_event_ids: Dict[DecisionId, int] = {} + self._pending_version_marker_event_ids: Dict[DecisionId, int] = {} self._mutable_side_effects: dict[str, MutableSideEffectState] = {} self.state_machines: OrderedDict[DecisionId, DecisionStateMachine] = ( OrderedDict() @@ -282,6 +287,29 @@ def record_mutable_side_effect( state.value = result return result + def version_marker_details(self, change_id: str) -> Payload | None: + """Return a preloaded Python Version marker's details, if one exists.""" + return self._recorded_marker_details.get( + marker_decision_id(VERSION_MARKER_NAME, change_id) + ) + + def has_pending_version_marker(self, change_id: str) -> bool: + """Whether a preloaded Version marker still awaits output-event routing.""" + return ( + marker_decision_id(VERSION_MARKER_NAME, change_id) + in self._pending_version_marker_event_ids + ) + + def record_version_marker(self, change_id: str, details: Payload) -> Payload: + """Record or replay a Version marker without consuming a sequence ID.""" + return self._record_marker( + decision.RecordMarkerDecisionAttributes( + marker_name=VERSION_MARKER_NAME, + details=details, + ), + context_id=change_id, + ) + # ----- Workflow API ----- def complete_workflow(self, decision: decision.Decision) -> None: if self._replaying: @@ -311,8 +339,17 @@ def _add_state_machine(self, state: DecisionStateMachine) -> None: # ----- History routing ----- + def preload_marker_event(self, event: history.HistoryEvent) -> None: + """Preload a marker before workflow code runs.""" + self._handle_history_event(event, preloaded_marker=True) + def handle_history_event(self, event: history.HistoryEvent) -> None: """Dispatch history event to typed handlers using the global transition map.""" + self._handle_history_event(event, preloaded_marker=False) + + def _handle_history_event( + self, event: history.HistoryEvent, *, preloaded_marker: bool + ) -> None: attr = event.WhichOneof("attributes") event_attributes = getattr(event, attr) @@ -325,7 +362,10 @@ def handle_history_event(self, event: history.HistoryEvent) -> None: decision_type = event_action.decision_type action = event_action.action if decision_type is DecisionType.MARKER: - self._index_marker_details(event_attributes) + self._index_marker_details(event_attributes, event.event_id) + self._update_version_marker_preload_state( + event_attributes, event.event_id, preloaded_marker + ) machine = self._state_machine_for_marker_event(event_attributes) if machine is None: return @@ -341,6 +381,23 @@ def handle_history_event(self, event: history.HistoryEvent) -> None: if action.event_id_is_alias: self.aliases[(decision_type, event.event_id)] = machine + def _update_version_marker_preload_state( + self, + attrs: history.MarkerRecordedEventAttributes, + event_id: int, + preloaded_marker: bool, + ) -> None: + if attrs.marker_name != VERSION_MARKER_NAME: + return + context_id = marker_context_id(attrs) + if context_id is None or not context_id: + return + marker_id = marker_decision_id(VERSION_MARKER_NAME, context_id) + if preloaded_marker: + self._pending_version_marker_event_ids[marker_id] = event_id + elif self._pending_version_marker_event_ids.get(marker_id) == event_id: + del self._pending_version_marker_event_ids[marker_id] + def _state_machine_for_event( self, event_id: int, @@ -401,7 +458,7 @@ def _state_machine_for_marker_event( return machine def _index_marker_details( - self, attrs: history.MarkerRecordedEventAttributes + self, attrs: history.MarkerRecordedEventAttributes, event_id: int ) -> None: """Store the user payload from a recorded marker event, keyed by its marker DecisionId. @@ -413,8 +470,26 @@ def _index_marker_details( return context_id = marker_context_id(attrs) if context_id is None: + if attrs.marker_name == VERSION_MARKER_NAME and has_marker_header(attrs): + raise FatalDecisionError( + "Version marker contains an invalid Python MarkerHeader" + ) return + if attrs.marker_name == VERSION_MARKER_NAME and not context_id: + raise FatalDecisionError( + "Version marker contains an invalid Python MarkerHeader" + ) marker_id = marker_decision_id(attrs.marker_name, context_id) + if attrs.marker_name == VERSION_MARKER_NAME: + previous_event_id = self._version_marker_event_ids.get(marker_id) + if previous_event_id is None: + self._version_marker_event_ids[marker_id] = event_id + elif previous_event_id != event_id: + raise FatalDecisionError( + f"Received duplicate Version marker for change_id {context_id!r}" + ) + else: + return details = Payload(data=attrs.details.data) self._recorded_marker_details[marker_id] = details mutable_info = mutable_side_effect_marker_info(attrs) @@ -439,7 +514,9 @@ def _start_execution(self, replaying: bool, outcomes: List[history.HistoryEvent] for event in outcomes: self._determinism_tracker.add_expectation(event) if event.HasField("marker_recorded_event_attributes"): - self._index_marker_details(event.marker_recorded_event_attributes) + self._index_marker_details( + event.marker_recorded_event_attributes, event.event_id + ) def _end_execution(self) -> None: if self._replaying: diff --git a/cadence/_internal/workflow/statemachine/marker_state_machine.py b/cadence/_internal/workflow/statemachine/marker_state_machine.py index 6dce8ba..33fb532 100644 --- a/cadence/_internal/workflow/statemachine/marker_state_machine.py +++ b/cadence/_internal/workflow/statemachine/marker_state_machine.py @@ -76,6 +76,14 @@ def marker_header( return None +def has_marker_header( + attrs: decision.RecordMarkerDecisionAttributes + | history.MarkerRecordedEventAttributes, +) -> bool: + """Return whether this marker attempts to use the Python MarkerHeader format.""" + return MARKER_HEADER_KEY in attrs.header.fields + + def marker_context_id( attrs: decision.RecordMarkerDecisionAttributes | history.MarkerRecordedEventAttributes, diff --git a/cadence/_internal/workflow/workflow_engine.py b/cadence/_internal/workflow/workflow_engine.py index 3417b3e..3327388 100644 --- a/cadence/_internal/workflow/workflow_engine.py +++ b/cadence/_internal/workflow/workflow_engine.py @@ -190,8 +190,7 @@ def _process_decision_events( "replay_mode": decision_events.replay, }, ) - # Process through state machines (DecisionsHelper now delegates to DecisionManager) - self._decision_manager.handle_history_event(marker_event) + self._decision_manager.preload_marker_event(marker_event) # Phase 2: Apply input events in history order. for event in decision_events.input: diff --git a/cadence/testing/_workflow_environment.py b/cadence/testing/_workflow_environment.py index 0f7513f..02e5be0 100644 --- a/cadence/testing/_workflow_environment.py +++ b/cadence/testing/_workflow_environment.py @@ -58,7 +58,10 @@ ) from cadence._internal.activity._definition import BaseDefinition -from cadence._internal.workflow.deterministic_event_loop import DeterministicEventLoop +from cadence._internal.workflow.deterministic_event_loop import ( + DeterministicEventLoop, + FatalDecisionError, +) from cadence._internal.workflow.workflow_instance import WorkflowInstance from cadence.activity import ActivityContext, ActivityInfo from cadence.api.v1.common_pb2 import Payload, WorkflowExecution @@ -71,6 +74,7 @@ ChildWorkflowFuture, ChildWorkflowOptions, ResultType, + VersioningOption, WorkflowContext, WorkflowDefinition, WorkflowInfo, @@ -165,6 +169,7 @@ def __init__(self, env: "TestWorkflowEnvironment", info: WorkflowInfo) -> None: self._env = env self._info = info self._mutable_side_effect_values: dict[str, Any] = {} + self._versions: dict[str, int] = {} def info(self) -> WorkflowInfo: return self._info @@ -245,6 +250,60 @@ def mutable_side_effect( return value return previous + def get_version( + self, + change_id: str, + min_supported: int, + max_supported: int, + *options: VersioningOption, + ) -> int: + if not isinstance(change_id, str) or not change_id: + raise ValueError("change_id must be a non-empty str") + for name, value in ( + ("min_supported", min_supported), + ("max_supported", max_supported), + ): + if isinstance(value, bool) or not isinstance(value, int): + raise ValueError(f"{name} must be an int") + if min_supported > max_supported: + raise ValueError("min_supported must not be greater than max_supported") + if change_id in self._versions: + version = self._versions[change_id] + if version < min_supported or version > max_supported: + raise FatalDecisionError( + f"cached version {version} for change_id {change_id!r} is outside " + f"the supported range [{min_supported}, {max_supported}]" + ) + return version + else: + custom_version: int | None = None + use_min_version = False + for option in options: + if not isinstance(option, VersioningOption): + raise ValueError( + "get_version options must be VersioningOption values" + ) + if option._kind == "version": + custom_version = option._version + elif option._kind == "min": + use_min_version = True + else: + raise ValueError("invalid get_version option") + version = ( + custom_version + if custom_version is not None + else min_supported + if use_min_version + else max_supported + ) + if version < min_supported or version > max_supported: + raise ValueError( + f"selected version {version} for change_id {change_id!r} is outside " + f"the supported range [{min_supported}, {max_supported}]" + ) + self._versions.setdefault(change_id, version) + return version + async def signal_child_workflow( self, child_workflow_id: str, diff --git a/cadence/workflow.py b/cadence/workflow.py index 87ca245..2fd0217 100644 --- a/cadence/workflow.py +++ b/cadence/workflow.py @@ -15,6 +15,7 @@ cast, Any, Optional, + Literal, Union, Unpack, Generic, @@ -33,6 +34,7 @@ _QUERY_TYPES_QUERY_NAME = "__query_types" ResultType = TypeVar("ResultType") +DEFAULT_VERSION = -1 class RetryPolicy(TypedDict, total=False): @@ -214,6 +216,52 @@ def mutable_side_effect( return WorkflowContext.get().mutable_side_effect(id, fn, result_type, updated) +@dataclass(frozen=True) +class VersioningOption: + """An opaque option value accepted by :func:`get_version`.""" + + _kind: Literal["min", "version"] + _version: int | None = None + + def __post_init__(self) -> None: + if self._kind == "min": + if self._version is not None: + raise ValueError("min version option must not specify a version") + return + if self._kind != "version": + raise ValueError("invalid versioning option kind") + if isinstance(self._version, bool) or not isinstance(self._version, int): + raise ValueError("version option must specify an int version") + + +def execute_with_version(version: int) -> VersioningOption: + """Select a particular version for a new ``get_version`` marker.""" + if isinstance(version, bool) or not isinstance(version, int): + raise ValueError("version must be an int") + return VersioningOption("version", version) + + +def execute_with_min_version() -> VersioningOption: + """Select the minimum supported version for a new ``get_version`` marker.""" + return VersioningOption("min") + + +def get_version( + change_id: str, + min_supported: int, + max_supported: int, + *options: VersioningOption, +) -> int: + """Return the deterministic version selected for ``change_id``. + + The first non-default value selected for a change is stored in a Version + marker. Replays return the recorded value and ignore selection options. + """ + return WorkflowContext.get().get_version( + change_id, min_supported, max_supported, *options + ) + + def is_cancel_requested() -> bool: return WorkflowContext.get().is_cancel_requested() @@ -649,6 +697,15 @@ def mutable_side_effect( updated: Callable[[ResultType, ResultType], bool], ) -> ResultType: ... + @abstractmethod + def get_version( + self, + change_id: str, + min_supported: int, + max_supported: int, + *options: VersioningOption, + ) -> int: ... + @abstractmethod def is_cancel_requested(self) -> bool: ... diff --git a/tests/cadence/_internal/workflow/test_context_versioning.py b/tests/cadence/_internal/workflow/test_context_versioning.py new file mode 100644 index 0000000..74b9a49 --- /dev/null +++ b/tests/cadence/_internal/workflow/test_context_versioning.py @@ -0,0 +1,541 @@ +import asyncio +from unittest.mock import MagicMock + +import pytest + +from cadence._internal.workflow.context import Context +from cadence._internal.workflow.deterministic_event_loop import FatalDecisionError +from cadence._internal.workflow.statemachine.decision_manager import DecisionManager +from cadence._internal.workflow.statemachine.marker_state_machine import ( + MARKER_HEADER_KEY, + VERSION_MARKER_NAME, + encode_marker_header, + marker_context_id, +) +from cadence.testing._workflow_environment import _InMemoryWorkflowContext +from cadence.api.v1 import decision, history +from cadence.api.v1.common_pb2 import Header, Payload +from cadence.data_converter import DefaultDataConverter +from cadence.workflow import ( + DEFAULT_VERSION, + VersioningOption, + WorkflowInfo, + execute_with_min_version, + execute_with_version, + get_version, +) + + +def _info() -> WorkflowInfo: + return WorkflowInfo( + workflow_type="Wf", + workflow_domain="domain", + workflow_id="wid", + workflow_run_id="rid", + workflow_task_list="tl", + data_converter=DefaultDataConverter(), + ) + + +def _context(*, replay: bool = False) -> tuple[Context, MagicMock]: + manager = MagicMock() + manager.version_marker_details.return_value = None + context = Context(_info(), manager) + context.set_replay_mode(replay) + return context, manager + + +def test_get_version_records_native_python_marker(): + context, manager = _context() + + assert context.get_version("change", 1, 2) == 2 + + manager.record_version_marker.assert_called_once_with( + "change", _info().data_converter.to_data([2]) + ) + + +def test_get_version_options_follow_go_precedence_and_cache(): + context, manager = _context() + + assert ( + context.get_version( + "change", + 1, + 5, + execute_with_min_version(), + execute_with_version(2), + execute_with_version(3), + ) + == 3 + ) + # Custom selection wins even when ExecuteWithMinVersion is applied last. + assert ( + context.get_version( + "change", 1, 5, execute_with_min_version(), execute_with_version(1) + ) + == 3 + ) + assert manager.record_version_marker.call_count == 1 + + +def test_get_version_execute_with_min_version_selects_minimum(): + context, _ = _context() + + assert context.get_version("change", 2, 5, execute_with_min_version()) == 2 + + +def test_get_version_default_version_does_not_record_a_marker(): + context, manager = _context() + + assert ( + context.get_version( + "change", DEFAULT_VERSION, 1, execute_with_version(DEFAULT_VERSION) + ) + == DEFAULT_VERSION + ) + manager.record_version_marker.assert_not_called() + + +def test_get_version_old_replay_without_marker_returns_default_and_emits_nothing(): + context, manager = _context(replay=True) + + assert context.get_version("change", DEFAULT_VERSION, 1) == DEFAULT_VERSION + manager.record_version_marker.assert_not_called() + + +def test_get_version_markerless_replay_rejects_an_unsupported_default_version(): + context, manager = _context(replay=True) + + with pytest.raises(FatalDecisionError, match="markerless replay version -1"): + context.get_version("change", 0, 1) + + manager.record_version_marker.assert_not_called() + + +def test_get_version_recorded_version_ignores_selection_options_and_is_revalidated(): + context, manager = _context(replay=True) + details = _info().data_converter.to_data([2]) + manager.version_marker_details.return_value = details + + assert context.get_version("change", 1, 3, execute_with_version(1)) == 2 + manager.record_version_marker.assert_called_once_with("change", details) + + with pytest.raises(FatalDecisionError, match="cached version 2"): + context.get_version("change", 3, 4, execute_with_version(3)) + + +@pytest.mark.parametrize( + ("change_id", "minimum", "maximum", "option"), + [ + ("", 1, 1, None), + ("change", True, 1, None), + ("change", 1, False, None), + ("change", 2, 1, None), + ("change", 1, 2, execute_with_version(3)), + ], +) +def test_get_version_validates_arguments( + change_id: str, minimum: int, maximum: int, option: object +): + context, _ = _context() + options = () if option is None else (option,) + + with pytest.raises(ValueError): + context.get_version(change_id, minimum, maximum, *options) + + +def test_execute_with_version_rejects_bool_and_non_int(): + with pytest.raises(ValueError): + execute_with_version(True) + with pytest.raises(ValueError): + execute_with_version("1") # type: ignore[arg-type] + + +def test_get_version_malformed_recorded_marker_is_a_fatal_decision_error(): + context, manager = _context(replay=True) + manager.version_marker_details.return_value = Payload(data=b'"not an int"') + + with pytest.raises(FatalDecisionError, match="Unable to decode Version marker"): + context.get_version("change", 1, 2) + + +@pytest.mark.parametrize( + "details", + [ + Payload(), + Payload(data=b"1 2"), + Payload(data=b"1 trailing"), + Payload(data=b"1x"), + ], +) +def test_get_version_rejects_invalid_default_converter_marker_details( + details: Payload, +): + context, manager = _context(replay=True) + manager.version_marker_details.return_value = details + + with pytest.raises(FatalDecisionError): + context.get_version("change", 1, 2) + + +def test_get_version_accepts_noncanonical_custom_converter_details(): + class RandomizedConverter: + def __init__(self) -> None: + self._sequence = 0 + + def from_data( + self, payload: Payload, type_hints: list[type | None] + ) -> list[int]: + return [int(payload.data.split(b":", maxsplit=1)[1])] + + def to_data(self, values: list[int]) -> Payload: + self._sequence += 1 + return Payload(data=f"{self._sequence}:{values[0]}".encode()) + + converter = RandomizedConverter() + manager = MagicMock() + details = Payload(data=b"99:2") + manager.version_marker_details.return_value = details + info = _info() + info = WorkflowInfo( + workflow_type=info.workflow_type, + workflow_domain=info.workflow_domain, + workflow_id=info.workflow_id, + workflow_run_id=info.workflow_run_id, + workflow_task_list=info.workflow_task_list, + data_converter=converter, + ) + context = Context(info, manager) + context.set_replay_mode(True) + + assert context.get_version("change", 1, 3) == 2 + manager.record_version_marker.assert_called_once_with("change", details) + + +def test_get_version_accepts_empty_details_from_a_custom_converter(): + class EmptyIntConverter: + def from_data( + self, payload: Payload, type_hints: list[type | None] + ) -> list[int]: + assert payload == Payload() + return [2] + + def to_data(self, values: list[int]) -> Payload: + return Payload() + + manager = MagicMock() + details = Payload() + manager.version_marker_details.return_value = details + info = _info() + context = Context( + WorkflowInfo( + workflow_type=info.workflow_type, + workflow_domain=info.workflow_domain, + workflow_id=info.workflow_id, + workflow_run_id=info.workflow_run_id, + workflow_task_list=info.workflow_task_list, + data_converter=EmptyIntConverter(), + ), + manager, + ) + context.set_replay_mode(True) + + assert context.get_version("change", 1, 3) == 2 + manager.record_version_marker.assert_called_once_with("change", details) + + +def test_get_version_rejects_executable_options_without_running_them(): + context, manager = _context() + calls: list[None] = [] + + def executable_option(_: object) -> None: + calls.append(None) + + with pytest.raises(ValueError, match="VersioningOption"): + context.get_version("change", 1, 2, executable_option) # type: ignore[arg-type] + + assert calls == [] + assert context.get_version("change", 1, 2) == 2 + manager.record_version_marker.assert_called_once() + + +@pytest.mark.parametrize( + "args", + [ + ("unknown", None), + ("min", 1), + ("version", None), + ("version", True), + ("version", "1"), + ], +) +def test_versioning_option_constructor_validates_invariants( + args: tuple[str, object], +): + with pytest.raises(ValueError): + VersioningOption(*args) # type: ignore[arg-type] + + +def test_in_memory_context_does_not_cache_a_failed_version_selection(): + context = _InMemoryWorkflowContext(MagicMock(), _info()) + + with pytest.raises(ValueError): + context.get_version("change", 1, 2, execute_with_version(3)) + + assert context.get_version("change", 1, 2) == 2 + + +def test_in_memory_context_rejects_cached_version_range_with_fatal_error(): + context = _InMemoryWorkflowContext(MagicMock(), _info()) + assert context.get_version("change", 1, 2) == 2 + + with pytest.raises(FatalDecisionError, match="cached version 2"): + context.get_version("change", 3, 4) + + +@pytest.mark.parametrize("change_id", [None, True, 1, b"change"]) +def test_production_context_requires_a_non_empty_string_change_id(change_id: object): + context, manager = _context() + + with pytest.raises(ValueError, match="non-empty str"): + context.get_version(change_id, 1, 2) # type: ignore[arg-type] + + manager.record_version_marker.assert_not_called() + + +@pytest.mark.parametrize("change_id", [None, True, 1, b"change"]) +def test_in_memory_context_requires_a_non_empty_string_change_id(change_id: object): + context = _InMemoryWorkflowContext(MagicMock(), _info()) + + with pytest.raises(ValueError, match="non-empty str"): + context.get_version(change_id, 1, 2) # type: ignore[arg-type] + + +def test_public_get_version_dispatches_through_context(): + context, _ = _context() + + with context._activate(): + assert get_version("change", 1, 2) == 2 + + +async def test_version_marker_has_stable_id_header_and_does_not_consume_sequence(): + manager = DecisionManager(asyncio.get_event_loop()) + details = DefaultDataConverter().to_data([7]) + + manager.record_version_marker("change", details) + timer = decision.StartTimerDecisionAttributes() + manager.start_timer(timer) + + pending = manager.collect_pending_decisions() + marker = pending[0].record_marker_decision_attributes + assert marker.marker_name == VERSION_MARKER_NAME + assert marker.details == details + assert marker_context_id(marker) == "change" + assert timer.timer_id == "0" + assert [key.id for key in manager.state_machines] == ["Version_change", "0"] + + +async def test_replay_preloads_python_version_marker_and_completes_its_state_machine(): + manager = DecisionManager(asyncio.get_event_loop()) + details = DefaultDataConverter().to_data([2]) + marker_event = history.HistoryEvent( + event_id=1, + marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( + marker_name=VERSION_MARKER_NAME, + details=details, + header=Header(fields={MARKER_HEADER_KEY: encode_marker_header("change")}), + ), + ) + context = Context(_info(), manager) + context.set_replay_mode(True) + + with manager.track_nondeterminism(True, []): + manager.preload_marker_event(marker_event) + assert context.get_version("change", 1, 3, execute_with_version(1)) == 2 + assert manager.collect_pending_decisions() == [ + decision.Decision( + record_marker_decision_attributes=decision.RecordMarkerDecisionAttributes( + marker_name=VERSION_MARKER_NAME, + details=details, + header=Header( + fields={MARKER_HEADER_KEY: encode_marker_header("change")} + ), + ) + ) + ] + manager.handle_history_event(marker_event) + assert manager.collect_pending_decisions() == [] + + +async def test_replay_does_not_recreate_consumed_version_marker_in_later_batch(): + manager = DecisionManager(asyncio.get_event_loop()) + details = DefaultDataConverter().to_data([2]) + marker_event = history.HistoryEvent( + event_id=1, + marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( + marker_name=VERSION_MARKER_NAME, + details=details, + header=Header(fields={MARKER_HEADER_KEY: encode_marker_header("change")}), + ), + ) + context = Context(_info(), manager) + context.set_replay_mode(True) + + # The original decision batch did not call get_version, so its marker is + # consumed without a state machine. + with manager.track_nondeterminism(True, []): + manager.preload_marker_event(marker_event) + manager.handle_history_event(marker_event) + assert manager.collect_pending_decisions() == [] + + # A moved get_version call still observes the recorded value, but must not + # recreate a decision for a marker whose output event is already consumed. + with manager.track_nondeterminism(True, []): + assert context.get_version("change", 1, 3) == 2 + assert manager.collect_pending_decisions() == [] + + +async def test_markerless_replay_default_is_replaced_by_later_version_marker(): + manager = DecisionManager(asyncio.get_event_loop()) + context = Context(_info(), manager) + context.set_replay_mode(True) + details = DefaultDataConverter().to_data([2]) + marker_event = history.HistoryEvent( + event_id=1, + marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( + marker_name=VERSION_MARKER_NAME, + details=details, + header=Header(fields={MARKER_HEADER_KEY: encode_marker_header("change")}), + ), + ) + + # First replay batch predates the Version marker. + with manager.track_nondeterminism(True, []): + assert context.get_version("change", DEFAULT_VERSION, 2) == DEFAULT_VERSION + assert context._versions == {"change": None} + + # A later batch exposes the marker. The provisional default must not mask it. + with manager.track_nondeterminism(True, []): + manager.preload_marker_event(marker_event) + assert context.get_version("change", 1, 2) == 2 + assert context._versions == {"change": 2} + assert manager.collect_pending_decisions() == [ + decision.Decision( + record_marker_decision_attributes=decision.RecordMarkerDecisionAttributes( + marker_name=VERSION_MARKER_NAME, + details=details, + header=Header( + fields={MARKER_HEADER_KEY: encode_marker_header("change")} + ), + ) + ) + ] + manager.handle_history_event(marker_event) + assert manager.collect_pending_decisions() == [] + + +async def test_replay_rejects_distinct_duplicate_version_markers(): + manager = DecisionManager(asyncio.get_event_loop()) + details = DefaultDataConverter().to_data([2]) + first_marker = history.HistoryEvent( + event_id=1, + marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( + marker_name=VERSION_MARKER_NAME, + details=details, + header=Header(fields={MARKER_HEADER_KEY: encode_marker_header("change")}), + ), + ) + second_marker = history.HistoryEvent( + event_id=2, + marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( + marker_name=VERSION_MARKER_NAME, + details=details, + header=Header(fields={MARKER_HEADER_KEY: encode_marker_header("change")}), + ), + ) + + with manager.track_nondeterminism(True, []): + manager.preload_marker_event(first_marker) + # Marker preloading and output routing both process this same event. + manager.handle_history_event(first_marker) + with pytest.raises(FatalDecisionError, match="duplicate Version marker"): + manager.preload_marker_event(second_marker) + + +async def test_version_markers_coexist_with_existing_markers_without_shifting_ids(): + manager = DecisionManager(asyncio.get_event_loop()) + details = DefaultDataConverter().to_data([2]) + version_event = history.HistoryEvent( + event_id=1, + marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( + marker_name=VERSION_MARKER_NAME, + details=details, + header=Header(fields={MARKER_HEADER_KEY: encode_marker_header("change")}), + ), + ) + side_effect_event = history.HistoryEvent( + event_id=2, + marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( + marker_name="SideEffect", + details=Payload(data=b"side-effect"), + header=Header(fields={MARKER_HEADER_KEY: encode_marker_header("0")}), + ), + ) + context = Context(_info(), manager) + context.set_replay_mode(True) + side_effect = decision.RecordMarkerDecisionAttributes( + marker_name="SideEffect", details=Payload(data=b"new-value") + ) + + with manager.track_nondeterminism(True, [side_effect_event]): + manager.preload_marker_event(version_event) + assert context.get_version("change", 1, 3) == 2 + manager.record_marker(side_effect) + assert marker_context_id(side_effect) == "0" + assert [item.get_id().id for item in manager.state_machines.values()] == [ + "Version_change", + "SideEffect_0", + ] + manager.handle_history_event(version_event) + manager.handle_history_event(side_effect_event) + + +async def test_replay_ignores_foreign_version_marker_format(): + manager = DecisionManager(asyncio.get_event_loop()) + context = Context(_info(), manager) + context.set_replay_mode(True) + foreign_marker = history.HistoryEvent( + event_id=1, + marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( + marker_name=VERSION_MARKER_NAME, + details=Payload(data=b'{"version": 2}'), + ), + ) + + with manager.track_nondeterminism(True, []): + manager.preload_marker_event(foreign_marker) + assert context.get_version("change", DEFAULT_VERSION, 2) == DEFAULT_VERSION + assert manager.collect_pending_decisions() == [] + + +@pytest.mark.parametrize( + "header_data", + [b"not-json", b"{}", b'{"context_id":""}'], +) +async def test_replay_rejects_malformed_python_version_marker_header( + header_data: bytes, +): + manager = DecisionManager(asyncio.get_event_loop()) + malformed_marker = history.HistoryEvent( + event_id=1, + marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( + marker_name=VERSION_MARKER_NAME, + details=DefaultDataConverter().to_data([2]), + header=Header(fields={MARKER_HEADER_KEY: Payload(data=header_data)}), + ), + ) + + with manager.track_nondeterminism(True, []): + with pytest.raises(FatalDecisionError, match="invalid Python MarkerHeader"): + manager.preload_marker_event(malformed_marker)