diff --git a/cadence/_internal/workflow/context.py b/cadence/_internal/workflow/context.py index 300d9446..abd12b40 100644 --- a/cadence/_internal/workflow/context.py +++ b/cadence/_internal/workflow/context.py @@ -9,7 +9,10 @@ from cadence._internal.workflow.child_workflow_future import ChildWorkflowFutureImpl from cadence._internal.workflow.deterministic_event_loop import DeterministicEventLoop -from cadence._internal.workflow.deterministic_event_loop import FatalDecisionError +from cadence._internal.workflow.deterministic_event_loop import ( + FatalDecisionError, + fail_decision, +) from cadence._internal.workflow.memo import memo_to_proto from cadence._internal.workflow.retry_policy import retry_policy_to_proto from cadence._internal.workflow.search_attributes import search_attributes_to_proto @@ -401,9 +404,11 @@ def _decode_recorded_version(self, change_id: str, details: Payload) -> int: try: return decode_version_marker_details(details) except ValueError as exc: - raise FatalDecisionError( + error = FatalDecisionError( f"Unable to decode Version marker for change_id {change_id!r}" - ) from exc + ) + error.__cause__ = exc + fail_decision(error) def set_replay_current_time(self, current_time: datetime) -> None: """Set the current replay timestamp.""" diff --git a/cadence/_internal/workflow/deterministic_event_loop.py b/cadence/_internal/workflow/deterministic_event_loop.py index 46dc7e9b..87824f28 100644 --- a/cadence/_internal/workflow/deterministic_event_loop.py +++ b/cadence/_internal/workflow/deterministic_event_loop.py @@ -1,11 +1,22 @@ +import inspect import traceback -from asyncio import AbstractEventLoop, Handle, TimerHandle, futures, tasks, Future, Task -from contextvars import Context +import weakref +from asyncio import ( + AbstractEventLoop, + CancelledError, + Handle, + TimerHandle, + futures, + tasks, + Future, + Task, +) +from contextvars import Context, copy_context import logging import collections import asyncio.events as events import threading -from typing import Callable, Any, TypeVar, Coroutine, Awaitable, Generator +from typing import Callable, Any, TypeVar, Coroutine, Awaitable, Generator, NoReturn from typing_extensions import Unpack, TypeVarTuple from cadence._internal.workflow.waiter import Waiter @@ -13,11 +24,37 @@ logger = logging.getLogger(__name__) -class FatalDecisionError(Exception): +class FatalDecisionError(BaseException): + """Fails the decision task rather than the workflow. + + Extends BaseException so that ``except Exception`` in workflow code doesn't + swallow it. + """ + def __init__(self, *args) -> None: super().__init__(*args) +class DecisionAbortedError(FatalDecisionError): + """Raised in workflow code once the decision has failed or the loop is closing. + + Its ``__cause__`` is the original failure, if any. The original is never raised + inside workflow code, so its traceback and context are reported unchanged. + """ + + +def fail_decision(error: FatalDecisionError) -> NoReturn: + """Mark the running decision as failed and raise error. + + The DeterministicEventLoop stops running workflow code and re-raises error, + so the decision task fails even if workflow code catches it. + """ + loop = events._get_running_loop() + if isinstance(loop, DeterministicEventLoop): + loop.fail_decision(error) + raise error + + _Ts = TypeVarTuple("_Ts") _T = TypeVar("_T") @@ -37,15 +74,45 @@ def __init__(self) -> None: self._waiters: list[Waiter] = [] self._stopping: bool = False self._closed: bool = False + self._decision_failure: FatalDecisionError | None = None + # Insertion-ordered so close() unwinds tasks deterministically. + self._tasks: weakref.WeakValueDictionary[int, _DeterministicTask[Any]] = ( + weakref.WeakValueDictionary() + ) + self._task_seq: int = 0 def run_until_yield(self): - """Run until stop() is called.""" + """Run until there is no ready work. + + Raises the FatalDecisionError passed to fail_decision(), if any. + """ self._run_forever_setup() try: - while self._ready: + while self._ready and self._decision_failure is None: self._run_once() finally: self._run_forever_cleanup() + if self._decision_failure is not None: + raise self._decision_failure + + def fail_decision(self, error: FatalDecisionError) -> None: + """Stop running workflow code and fail the decision with error.""" + if self._decision_failure is None: + self._decision_failure = error + + def abort_if_decision_failed(self) -> None: + """Raise a DecisionAbortedError in workflow code if the decision has failed + or the loop is closed.""" + if self._closed or self._decision_failure is not None: + raise self._abort_error() + + def _abort_error(self) -> DecisionAbortedError: + failure = self._decision_failure + if failure is None: + return DecisionAbortedError("workflow destroyed") + error = DecisionAbortedError("decision already failed") + error.__cause__ = failure + return error # Event Loop APIs def call_soon( @@ -78,10 +145,12 @@ def run_forever(self) -> None: try: while True: self._run_once() - if self._stopping: + if self._stopping or self._decision_failure is not None: break finally: self._run_forever_cleanup() + if self._decision_failure is not None: + raise self._decision_failure def run_until_complete( self, future: Generator[Any, None, _T] | Awaitable[_T] @@ -126,19 +195,23 @@ def create_task( Return a task object. """ - self._check_closed() + if self._closed or self._decision_failure is not None: + # The loop won't run again, so the task could never start. + if inspect.iscoroutine(coro): + coro.close() + raise self._abort_error() # NOTE: eager_start is not supported for deterministic event loop - if kwargs.get("eager_start", False): + if kwargs.pop("eager_start", False): + if inspect.iscoroutine(coro): + coro.close() raise RuntimeError( "eager_start in create_task is not supported for deterministic event loop" ) - task = tasks.Task(coro, loop=self, **kwargs) - # We intentionally destroy pending tasks when shutting down the event loop. - # If our asyncio implementation supports it, disable the logs - if hasattr(task, "_log_destroy_pending"): - setattr(task, "_log_destroy_pending", False) + task = _DeterministicTask(coro, loop=self, **kwargs) + self._tasks[self._task_seq] = task + self._task_seq += 1 return task def create_future(self) -> Future[Any]: @@ -158,6 +231,8 @@ def _run_once(self) -> None: if handle._cancelled: continue handle._run() + if self._decision_failure is not None: + return # Poll waiters; only stop early if settling one schedules new work, # so remaining waiters are not skipped. @@ -165,7 +240,10 @@ def _run_once(self) -> None: while i < len(self._waiters): w = self._waiters[i] ready_before = len(self._ready) - if w.poll(): + settled = w.poll() + if self._decision_failure is not None: + return + if settled: del self._waiters[i] if len(self._ready) > ready_before: return @@ -175,6 +253,9 @@ def _run_once(self) -> None: def _run_forever_setup(self) -> None: self._check_closed() self._check_running() + self._set_running() + + def _set_running(self) -> None: self._thread_id = threading.get_ident() events._set_running_loop(self) @@ -188,7 +269,7 @@ def stop(self) -> None: def _check_closed(self) -> None: if self._closed: - raise RuntimeError("Event loop is closed") + raise self._abort_error() def _check_running(self) -> None: if self.is_running(): @@ -202,8 +283,12 @@ def is_running(self) -> bool: return self._thread_id is not None def close(self) -> None: - """Close the event loop. - The event loop must not be running. + """Close the loop and unwind every pending coroutine. + + The event loop must not be running. Coroutines are unwound directly rather + than by running the loop, so each finally block runs once and no other + workflow code is scheduled. Once closed, interacting with the loop raises + DecisionAbortedError. """ if self.is_running(): raise RuntimeError("Cannot close a running event loop") @@ -211,9 +296,21 @@ def close(self) -> None: return if self._debug: logger.debug("Close %r", self) + self._check_running() self._closed = True - self._ready.clear() - self._waiters.clear() + # Mark the loop as running so get_running_loop() works in finally blocks. + self._set_running() + try: + # create_task() refuses new tasks once the loop is closed, so finally + # blocks can't add to this snapshot. + for task in reversed(list(self._tasks.values())): + task.destroy(self._abort_error) + finally: + self._run_forever_cleanup() + # Callbacks scheduled while unwinding are dropped on purpose: every + # task they would wake has already been unwound. + self._ready.clear() + self._waiters.clear() def is_closed(self) -> bool: """Returns True if the event loop was closed.""" @@ -547,6 +644,78 @@ async def connect_accepted_socket( ) +# Bounds how long workflow code that keeps catching the error can delay close(). +_MAX_DESTROY_THROWS = 100 + + +class _DeterministicTask(Task[_T]): + """A Task that keeps its Context so close() can unwind it in that Context.""" + + def __init__( + self, + coro: Generator[Any, None, _T] | Coroutine[Any, Any, _T], + *, + loop: AbstractEventLoop, + name: str | None = None, + context: Context | None = None, + ) -> None: + if context is None: + context = copy_context() + super().__init__(coro, loop=loop, name=name, context=context) + self._destroy_context = context + # close() unwinds coroutines directly, so tasks are never marked done. + # Disable the "Task was destroyed but it is pending!" log for them. + if hasattr(self, "_log_destroy_pending"): + setattr(self, "_log_destroy_pending", False) + + def destroy(self, abort_error: Callable[[], DecisionAbortedError]) -> None: + if self.done(): + if isinstance(getattr(self, "_exception", None), FatalDecisionError): + # Already surfaced as the decision failure, don't log it again + # as "Task exception was never retrieved". Other exceptions are + # left alone so unretrieved user errors are still reported. + self._log_traceback = False + return + coro = self.get_coro() + if not inspect.iscoroutine(coro): + return + state = inspect.getcoroutinestate(coro) + if state == inspect.CORO_CREATED: + # Never started: closing runs no workflow code. + coro.close() + return + if state != inspect.CORO_SUSPENDED: + return + # Each async context manager whose __aexit__ awaits (e.g. a TaskGroup waiting on + # its children) yields once while handling the error, so throw again at each new + # await point until the coroutine finishes. close() can't do this: awaiting while + # handling GeneratorExit fails and leaves the coroutine for the garbage collector. + for _ in range(_MAX_DESTROY_THROWS): + try: + self._destroy_context.run(coro.throw, abort_error()) + except (StopIteration, FatalDecisionError): + return + except BaseExceptionGroup as e: + if e.split(FatalDecisionError)[1] is not None: + logger.warning( + "Workflow code raised while being destroyed", exc_info=True + ) + return + except (Exception, CancelledError): + logger.warning( + "Workflow code raised while being destroyed", exc_info=True + ) + return + try: + self._destroy_context.run(coro.close) + except (Exception, CancelledError): + logger.error( + "Workflow coroutine could not be destroyed, it will be finalized by " + "the garbage collector", + exc_info=True, + ) + + def _run_until_complete_cb(fut: Future[Any]) -> None: if not fut.cancelled(): exc = fut.exception() diff --git a/cadence/_internal/workflow/statemachine/decision_manager.py b/cadence/_internal/workflow/statemachine/decision_manager.py index 5c8b6d15..b5b0d447 100644 --- a/cadence/_internal/workflow/statemachine/decision_manager.py +++ b/cadence/_internal/workflow/statemachine/decision_manager.py @@ -46,7 +46,11 @@ ) 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.deterministic_event_loop import ( + DeterministicEventLoop, + FatalDecisionError, + fail_decision, +) from cadence._internal.workflow.statemachine.signal_external_workflow_state_machine import ( signal_external_events, SignalExternalWorkflowStateMachine, @@ -125,7 +129,7 @@ class DecisionManager: } ) - def __init__(self, event_loop: asyncio.AbstractEventLoop): + def __init__(self, event_loop: DeterministicEventLoop): self._event_loop = event_loop self._id_counter = 0 self._determinism_tracker = DeterminismTracker() @@ -322,8 +326,10 @@ def _get_or_create_version_marker( existing = self.state_machines.get(marker_id) if existing is not None: if not isinstance(existing, MarkerStateMachine): - raise FatalDecisionError( - f"Unexpected state machine for Version marker {change_id!r}" + fail_decision( + FatalDecisionError( + f"Unexpected state machine for Version marker {change_id!r}" + ) ) return existing @@ -361,6 +367,8 @@ def _get_machine(self, decision_id: DecisionId) -> DecisionStateMachine: return machine def _add_state_machine(self, state: DecisionStateMachine) -> None: + # No new decisions once the decision has failed or the workflow is destroyed. + self._event_loop.abort_if_decision_failed() decision_id = state.get_id() if decision_id in self.state_machines: raise ValueError(f"Received duplicate decision: {decision_id}") @@ -466,8 +474,10 @@ def _state_machine_for_marker_event( and machine.was_recorded and machine.get_result().data != event_attributes.details.data ): - raise FatalDecisionError( - f"Received conflicting Version marker for change_id {context_id!r}" + fail_decision( + FatalDecisionError( + f"Received conflicting Version marker for change_id {context_id!r}" + ) ) if machine is None: if event_attributes.marker_name == VERSION_MARKER_NAME: @@ -504,13 +514,17 @@ def _index_marker_details( 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" + fail_decision( + 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" + fail_decision( + FatalDecisionError( + "Version marker contains an invalid Python MarkerHeader" + ) ) if attrs.marker_name == VERSION_MARKER_NAME: return @@ -590,6 +604,7 @@ def request_cancel(message: str | None = None) -> bool: def _request_cancel( self, decision_id: DecisionId, message: str | None = None ) -> bool: + self._event_loop.abort_if_decision_failed() machine = self._get_machine(decision_id) if machine.request_cancel(message): if self._replaying: diff --git a/cadence/_internal/workflow/statemachine/nondeterminism.py b/cadence/_internal/workflow/statemachine/nondeterminism.py index d65057c9..780a9000 100644 --- a/cadence/_internal/workflow/statemachine/nondeterminism.py +++ b/cadence/_internal/workflow/statemachine/nondeterminism.py @@ -2,7 +2,10 @@ from functools import singledispatch from typing import Any, OrderedDict, Dict, List, Never -from cadence._internal.workflow.deterministic_event_loop import FatalDecisionError +from cadence._internal.workflow.deterministic_event_loop import ( + FatalDecisionError, + fail_decision, +) from cadence._internal.workflow.statemachine.cancellation import ( is_immediate_cancel, from_marker, @@ -113,8 +116,10 @@ def _add_expectations( self, decision_id: DecisionId, to_expect: List[Expectation] ) -> None: if decision_id in self._expectations: - raise FatalDecisionError( - f"Received duplicate expectation for {decision_id}: {self._expectations[decision_id]} and {to_expect}" + fail_decision( + FatalDecisionError( + f"Received duplicate expectation for {decision_id}: {self._expectations[decision_id]} and {to_expect}" + ) ) self._expectations[decision_id] = to_expect @@ -187,7 +192,7 @@ def _validate_expectation(self, actual: Expectation) -> Expectation: def _fail(self, expected: Expectation | None, actual: Expectation | None) -> Never: self._failed = True - raise NonDeterminismError(expected, actual) + fail_decision(NonDeterminismError(expected, actual)) def _next_expectation(self) -> Expectation: _, upcoming = next(iter(self._expectations.items())) @@ -202,7 +207,7 @@ def complete_replay(self) -> None: def record_immediate_cancel(attrs: Any) -> decision.Decision: expectation = to_expectation(attrs) if expectation is None: - raise FatalDecisionError(f"Unable to handle type: {attrs} ") + fail_decision(FatalDecisionError(f"Unable to handle type: {attrs} ")) return decision.Decision( record_marker_decision_attributes=to_marker( expectation.decision_id, expectation.properties diff --git a/cadence/_internal/workflow/versioning.py b/cadence/_internal/workflow/versioning.py index 2dd714ce..6e8e6e19 100644 --- a/cadence/_internal/workflow/versioning.py +++ b/cadence/_internal/workflow/versioning.py @@ -3,7 +3,10 @@ from msgspec import DecodeError, json -from cadence._internal.workflow.deterministic_event_loop import FatalDecisionError +from cadence._internal.workflow.deterministic_event_loop import ( + FatalDecisionError, + fail_decision, +) from cadence.api.v1.common_pb2 import Payload from cadence.workflow import ( CADENCE_CHANGE_VERSION_SEARCH_ATTRIBUTE, @@ -55,9 +58,11 @@ def validate_resolved_version( max_supported: int, ) -> None: if version < min_supported or version > max_supported: - raise FatalDecisionError( - f"version {version} for change_id {change_id!r} is outside " - f"the supported range [{min_supported}, {max_supported}]" + fail_decision( + FatalDecisionError( + f"version {version} for change_id {change_id!r} is outside " + f"the supported range [{min_supported}, {max_supported}]" + ) ) diff --git a/cadence/_internal/workflow/workflow_engine.py b/cadence/_internal/workflow/workflow_engine.py index 1b5d5ef6..bbeb4f35 100644 --- a/cadence/_internal/workflow/workflow_engine.py +++ b/cadence/_internal/workflow/workflow_engine.py @@ -128,7 +128,7 @@ def process_decision( return DecisionResult(decisions=decisions, query_result=None) # TODO: reevaluate if this is needed to log error here or in the caller - except Exception as e: + except (Exception, FatalDecisionError) as e: # Log decision task failure with full context (matches Java ReplayDecisionTaskHandler) logger.error( "Decision task processing failed", @@ -164,6 +164,12 @@ def _execute_query(self, query: WorkflowQuery) -> DecisionResult: def is_done(self) -> bool: return self._workflow_instance.is_done() + def destroy(self) -> None: + """Close the event loop, shutting down all Workflow coroutines""" + with _activate_workflow_context(self._workflow_context): + with extract_headers(self._context_propagators, self._headers): + self._event_loop.close() + def _process_decision_events( self, ctx: Context, @@ -258,7 +264,7 @@ def _maybe_complete_workflow(self) -> Optional[Decision]: continue_as_new_workflow_execution_decision_attributes=attrs, ) except ExceptionGroup as e: - if e.subgroup((InvalidStateError, FatalDecisionError)): + if e.subgroup(InvalidStateError): raise failure = _failure_from_exception(e) diff --git a/cadence/testing/_workflow_environment.py b/cadence/testing/_workflow_environment.py index 47b00f32..e95ad043 100644 --- a/cadence/testing/_workflow_environment.py +++ b/cadence/testing/_workflow_environment.py @@ -63,7 +63,10 @@ from cadence._internal.activity._definition import BaseDefinition from cadence._internal.workflow.child_workflow_future import ChildWorkflowFutureImpl -from cadence._internal.workflow.deterministic_event_loop import DeterministicEventLoop +from cadence._internal.workflow.deterministic_event_loop import ( + DeterministicEventLoop, + FatalDecisionError, +) from cadence._internal.workflow.search_attributes import search_attributes_to_proto from cadence._internal.workflow.versioning import ( change_version_search_attributes, @@ -424,6 +427,9 @@ def run_start( self._run_until_settled() def run_signal(self, name: str, payload: Payload) -> None: + if isinstance(self.error, FatalDecisionError): + # The loop refuses new tasks once the decision has failed. + return with self._driving(): self._dispatch_signal(name, payload) self._run_until_settled() @@ -438,6 +444,10 @@ def _dispatch_signal(self, name: str, payload: Payload) -> None: args = signal_def.params_from_payload(self._data_converter, payload) self._instance.handle_signal(signal_def, args) + def destroy(self) -> None: + with self._driving(): + self._loop.close() + @contextmanager def _driving(self) -> Iterator[None]: # Mark this execution as the one currently being driven so timers @@ -460,16 +470,21 @@ def _run_until_settled(self) -> None: again. We stop once the workflow completes or is blocked on something other than a timer (e.g. an unsatisfied ``wait_condition``). """ - self._instance.run_until_yield() - while not self._instance.is_done() and self._timers: - deadline, _seq, future = heapq.heappop(self._timers) - if future.done(): - # Timer was cancelled (or already resolved); a cancelled timer - # does not consume virtual time, so don't advance the clock. - continue - self._env._advance_clock_to(deadline) - future.set_result(None) + try: self._instance.run_until_yield() + while not self._instance.is_done() and self._timers: + deadline, _seq, future = heapq.heappop(self._timers) + if future.done(): + # Timer was cancelled (or already resolved); a cancelled timer + # does not consume virtual time, so don't advance the clock. + continue + self._env._advance_clock_to(deadline) + future.set_result(None) + self._instance.run_until_yield() + except FatalDecisionError as exc: + self.completed = True + self.error = exc + return self._collect_outcome() def run_query(self, query_type: str, args: Payload) -> Payload: @@ -623,6 +638,7 @@ def __init__( self._workflow_interceptor_factory = workflow_interceptor_factory or _identity self._activity_mocks: dict[str, Union[_ValueMock, _FnMock]] = {} self._executions: dict[str, _Execution] = {} + self._closed = False self._last_execution: Optional[_Execution] = None # Signals emitted via signal_external_workflow/signal_child_workflow, # delivered after the current decision completes. @@ -705,6 +721,11 @@ def get_workflow_error( return self._get_execution(workflow_id).error def close(self) -> None: + if self._closed: + return + self._closed = True + for execution in self._executions.values(): + self._executor.submit(execution.destroy).result() self._executor.shutdown(wait=False) def __enter__(self) -> "TestWorkflowEnvironment": diff --git a/cadence/worker/_base_task_handler.py b/cadence/worker/_base_task_handler.py index c4ea1772..768d8d94 100644 --- a/cadence/worker/_base_task_handler.py +++ b/cadence/worker/_base_task_handler.py @@ -2,6 +2,7 @@ from abc import ABC, abstractmethod from typing import TypeVar, Generic +from cadence._internal.workflow.deterministic_event_loop import FatalDecisionError from cadence.metrics import MetricsEmitter, NoOpMetricsEmitter logger = logging.getLogger(__name__) @@ -41,7 +42,7 @@ async def handle_task(self, task: T) -> None: # Handle the task implementation await self._handle_task_implementation(task) - except Exception as e: + except (Exception, FatalDecisionError) as e: logger.exception(f"Error handling task: {e}") await self.handle_task_failure(task, e) @@ -56,7 +57,7 @@ async def _handle_task_implementation(self, task: T) -> None: pass @abstractmethod - async def handle_task_failure(self, task: T, error: Exception) -> None: + async def handle_task_failure(self, task: T, error: BaseException) -> None: """ Handle task processing failure. diff --git a/cadence/worker/_decision_task_handler.py b/cadence/worker/_decision_task_handler.py index 1f68adc5..93bb7417 100644 --- a/cadence/worker/_decision_task_handler.py +++ b/cadence/worker/_decision_task_handler.py @@ -5,6 +5,7 @@ import logging from typing import Optional, Sequence +from cadence._internal.workflow.deterministic_event_loop import FatalDecisionError from cadence._internal.workflow.history_event_iterator import iterate_history_events from cadence._internal.context import header_to_dict from cadence._internal.workflow.memo import memo_from_proto @@ -17,7 +18,9 @@ RespondDecisionTaskFailedRequest, RespondQueryTaskCompletedRequest, ) +from cadence.api.v1.history_pb2 import HistoryEvent from cadence.api.v1.query_pb2 import ( + WorkflowQuery, WorkflowQueryResult, QUERY_RESULT_TYPE_FAILED, ) @@ -63,6 +66,19 @@ } +def _process_and_destroy( + engine: WorkflowEngine, + events: list[HistoryEvent], + query: Optional[WorkflowQuery], +) -> DecisionResult: + # Engines aren't cached yet, so destroy after every task. + # With a sticky cache, destroy on eviction or failure instead. + try: + return engine.process_decision(events, query) + finally: + engine.destroy() + + class DecisionTaskHandler(BaseTaskHandler[PollForDecisionTaskResponse]): """ Task handler for processing decision tasks. @@ -212,11 +228,12 @@ async def _handle_task_implementation( try: decision_result = await asyncio.get_running_loop().run_in_executor( self._executor, - workflow_engine.process_decision, + _process_and_destroy, + workflow_engine, workflow_events, task.query if is_query_task else None, ) - except Exception: + except (Exception, FatalDecisionError): emitter.counter(DECISION_EXECUTION_FAILED_COUNTER) emitter.counter(DECISION_TASK_PANIC_COUNTER) raise @@ -251,7 +268,7 @@ async def _handle_task_implementation( ) async def handle_task_failure( - self, task: PollForDecisionTaskResponse, error: Exception + self, task: PollForDecisionTaskResponse, error: BaseException ) -> None: """ Handle decision task processing failure. diff --git a/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py b/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py index 6c16d446..53453c17 100644 --- a/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py +++ b/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py @@ -1,4 +1,4 @@ -import asyncio +from cadence._internal.workflow.deterministic_event_loop import DeterministicEventLoop from asyncio import CancelledError import pytest @@ -24,7 +24,7 @@ async def test_activity_dispatch(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) activity_result = decisions.schedule_activity( decision.ScheduleActivityTaskDecisionAttributes() @@ -38,7 +38,7 @@ async def test_activity_dispatch(): async def test_simple_cancellation(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) activity_result = decisions.schedule_activity( decision.ScheduleActivityTaskDecisionAttributes() @@ -50,7 +50,7 @@ async def test_simple_cancellation(): async def test_cancellation_not_immediate(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) activity_result = decisions.schedule_activity( decision.ScheduleActivityTaskDecisionAttributes() @@ -64,7 +64,7 @@ async def test_cancellation_not_immediate(): async def test_cancellation_with_message_cancels_future_immediately(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) activity_result = decisions.schedule_activity( decision.ScheduleActivityTaskDecisionAttributes() @@ -80,7 +80,7 @@ async def test_cancellation_with_message_cancels_future_immediately(): async def test_cancellation_completed(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) activity_result = decisions.schedule_activity( decision.ScheduleActivityTaskDecisionAttributes() @@ -111,7 +111,7 @@ async def test_cancellation_completed(): async def test_collect_decisions(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) activity1 = decisions.schedule_activity( decision.ScheduleActivityTaskDecisionAttributes() @@ -138,7 +138,7 @@ async def test_collect_decisions(): async def test_collect_decisions_ignore_empty(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) _ = decisions.schedule_activity(decision.ScheduleActivityTaskDecisionAttributes()) decisions.handle_history_event(activity_scheduled(1, "0")) @@ -148,7 +148,7 @@ async def test_collect_decisions_ignore_empty(): async def test_collect_pending_decisions_timer_cancel_before_complete(): """User-initiated timer cancel must still emit CancelTimer when workflow completes.""" - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) timer_fut = decisions.start_timer(decision.StartTimerDecisionAttributes()) decisions.handle_history_event( history.HistoryEvent( @@ -177,7 +177,7 @@ async def test_collect_pending_decisions_timer_cancel_before_complete(): async def test_collection_decisions_reordering(): # Decisions should be emitted in the order that they happened within the workflow - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) activity1 = decisions.schedule_activity( decision.ScheduleActivityTaskDecisionAttributes() @@ -223,7 +223,7 @@ async def test_collection_decisions_reordering(): async def test_record_marker_collects_in_workflow_order(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) marker_attrs = decision.RecordMarkerDecisionAttributes( marker_name="SideEffect", details=Payload(data=b"marker") ) @@ -249,7 +249,7 @@ async def test_record_marker_collects_in_workflow_order(): async def test_record_marker_history_clears_pending_decision(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) marker_attrs = decision.RecordMarkerDecisionAttributes( marker_name="SideEffect", details=Payload(data=b"marker") ) @@ -264,7 +264,7 @@ async def test_record_marker_history_clears_pending_decision(): async def test_replayed_marker_returns_recorded_details(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) recorded = marker_recorded( 1, "SideEffect", Payload(data=b"history-value"), context_id="0" ) @@ -280,7 +280,7 @@ async def test_replayed_marker_returns_recorded_details(): async def test_replayed_marker_name_mismatch_is_nondeterministic(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) recorded = marker_recorded( 1, "SideEffect", Payload(data=b"history-value"), context_id="0" ) @@ -293,7 +293,7 @@ async def test_replayed_marker_name_mismatch_is_nondeterministic(): async def test_replayed_marker_order_mismatch_is_nondeterministic(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) first = marker_recorded(1, "SideEffect", Payload(data=b"first"), context_id="0") second = marker_recorded( 2, "LocalActivity", Payload(data=b"second"), context_id="1" @@ -309,7 +309,7 @@ async def test_replayed_marker_order_mismatch_is_nondeterministic(): async def test_replayed_marker_missing_from_workflow_is_nondeterministic(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) recorded = marker_recorded( 1, "SideEffect", Payload(data=b"history-value"), context_id="0" ) @@ -320,7 +320,7 @@ async def test_replayed_marker_missing_from_workflow_is_nondeterministic(): async def test_replayed_marker_without_context_id_is_nondeterministic(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) # Raw details with no length-prefix encoding — treated as produced by another SDK. recorded = marker_recorded(1, "SideEffect", Payload(data=b"history-value")) @@ -332,7 +332,7 @@ async def test_replayed_marker_without_context_id_is_nondeterministic(): async def test_record_marker_ids_are_sequential(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) decisions.record_marker( decision.RecordMarkerDecisionAttributes(marker_name="SideEffect") @@ -347,7 +347,7 @@ async def test_record_marker_ids_are_sequential(): async def test_record_marker_requires_marker_name(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) with pytest.raises(ValueError, match="marker_name is required"): decisions.record_marker(decision.RecordMarkerDecisionAttributes()) @@ -356,7 +356,7 @@ async def test_record_marker_requires_marker_name(): async def test_mutable_side_effect_records_metadata_and_caches_value(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) access_count, stored, has_history_update = decisions.mutable_side_effect_value( "config" ) @@ -375,7 +375,7 @@ async def test_mutable_side_effect_records_metadata_and_caches_value(): async def test_mutable_side_effect_does_not_shift_other_decision_ids_on_replay(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) recorded = marker_recorded( 1, "MutableSideEffect", @@ -395,7 +395,7 @@ async def test_mutable_side_effect_does_not_shift_other_decision_ids_on_replay() async def test_mutable_side_effect_replay_applies_marker_at_matching_access_count(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) recorded = marker_recorded( 1, "MutableSideEffect", @@ -425,7 +425,7 @@ async def test_cancel_marker_is_not_logged_as_unknown_marker_name(caplog): # The immediate-cancellation marker (cancellation.py) is Python-specific and encodes # a JSON object in Details, not a [context_id, user_data] pair. It must not be mistaken # for an unrecognized marker_name and warned about. - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) cancel_marker = history.HistoryEvent( event_id=1, marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( @@ -444,7 +444,7 @@ async def test_version_marker_is_not_flagged_as_nondeterministic(): # Version markers are exempt from non-determinism tracking. # History that includes a Version marker must replay without error, and the # Version expectation is never added so complete_replay doesn't fail either. - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) version_recorded = marker_recorded( 1, "Version", Payload(data=b"v1"), context_id="0" ) @@ -465,7 +465,7 @@ async def test_version_marker_is_not_flagged_as_nondeterministic(): async def test_replayed_marker_details_come_from_history_not_current_value(): # Ensure the historical value is returned even when the current call passes different data. - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) recorded = marker_recorded( 1, "SideEffect", Payload(data=b"history-value"), context_id="0" ) @@ -481,7 +481,7 @@ async def test_replayed_marker_details_come_from_history_not_current_value(): async def test_child_workflow_dispatch(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) execution, result = decisions.schedule_child_workflow( decision.StartChildWorkflowExecutionDecisionAttributes( @@ -508,7 +508,7 @@ async def test_child_workflow_dispatch(): async def test_schedule_child_workflow_generates_workflow_id(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) attrs = decision.StartChildWorkflowExecutionDecisionAttributes( workflow_id="", @@ -532,7 +532,7 @@ async def test_schedule_child_workflow_generates_workflow_id(): async def test_child_workflow_initiation_failed_dispatch(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) execution, result = decisions.schedule_child_workflow( decision.StartChildWorkflowExecutionDecisionAttributes( @@ -561,7 +561,7 @@ async def test_child_workflow_initiation_failed_dispatch(): async def test_child_workflow_cancel_dispatch(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) _execution, result = decisions.schedule_child_workflow( decision.StartChildWorkflowExecutionDecisionAttributes( @@ -605,7 +605,7 @@ async def test_child_workflow_cancel_dispatch(): async def test_child_workflow_errors_are_child_workflow_error(): """All child workflow errors share a common base class.""" - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) _execution, result = decisions.schedule_child_workflow( decision.StartChildWorkflowExecutionDecisionAttributes( @@ -743,7 +743,7 @@ def marker_recorded( async def test_signal_external_workflow_creates_machine(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) future = decisions.signal_external_workflow( decision.SignalExternalWorkflowExecutionDecisionAttributes( @@ -764,7 +764,7 @@ async def test_signal_external_workflow_creates_machine(): async def test_signal_external_workflow_sets_control(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) attrs = decision.SignalExternalWorkflowExecutionDecisionAttributes( domain="test-domain", @@ -778,7 +778,7 @@ async def test_signal_external_workflow_sets_control(): async def test_signal_external_workflow_multiple_signals(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) future1 = decisions.signal_external_workflow( decision.SignalExternalWorkflowExecutionDecisionAttributes( @@ -809,7 +809,7 @@ async def test_signal_external_workflow_multiple_signals(): async def test_signal_external_workflow_dispatch(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) future = decisions.signal_external_workflow( decision.SignalExternalWorkflowExecutionDecisionAttributes( @@ -828,7 +828,7 @@ async def test_signal_external_workflow_dispatch(): async def test_signal_external_workflow_failed_dispatch(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) future = decisions.signal_external_workflow( decision.SignalExternalWorkflowExecutionDecisionAttributes( @@ -894,7 +894,7 @@ def _upsert_recorded(event_id: int, fields: dict[str, bytes]) -> history.History async def test_upsert_search_attributes_collects_decision(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) attrs = _upsert_attrs({"CustomIntField": b"1"}) decisions.upsert_search_attributes(attrs) @@ -906,7 +906,7 @@ async def test_upsert_search_attributes_collects_decision(): async def test_upsert_search_attributes_history_clears_pending_decision(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) attrs = _upsert_attrs({"CustomIntField": b"1"}) decisions.upsert_search_attributes(attrs) @@ -916,7 +916,7 @@ async def test_upsert_search_attributes_history_clears_pending_decision(): async def test_upsert_search_attributes_multiple_match_in_order(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) first = _upsert_attrs({"CustomIntField": b"1"}) second = _upsert_attrs({"CustomKeywordField": b'"seattle"'}) @@ -935,7 +935,7 @@ async def test_upsert_search_attributes_multiple_match_in_order(): async def test_replayed_upsert_matches_history(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) recorded = _upsert_recorded(1, {"CustomIntField": b"1"}) with decisions.track_nondeterminism(True, [recorded]): @@ -943,7 +943,7 @@ async def test_replayed_upsert_matches_history(): async def test_replayed_upsert_is_not_checked_for_nondeterminism(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) recorded = _upsert_recorded(1, {"CustomIntField": b"1"}) with decisions.track_nondeterminism(True, [recorded]): @@ -951,7 +951,7 @@ async def test_replayed_upsert_is_not_checked_for_nondeterminism(): async def test_upsert_history_without_requested_machine_is_ignored(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) decisions.handle_history_event(_upsert_recorded(1, {"CustomIntField": b"1"})) diff --git a/tests/cadence/_internal/workflow/test_context_versioning.py b/tests/cadence/_internal/workflow/test_context_versioning.py index 297905ad..0cb5cee9 100644 --- a/tests/cadence/_internal/workflow/test_context_versioning.py +++ b/tests/cadence/_internal/workflow/test_context_versioning.py @@ -1,10 +1,12 @@ -import asyncio from unittest.mock import MagicMock import pytest from cadence._internal.workflow.context import Context, _activate_workflow_context -from cadence._internal.workflow.deterministic_event_loop import FatalDecisionError +from cadence._internal.workflow.deterministic_event_loop import ( + DeterministicEventLoop, + FatalDecisionError, +) from cadence._internal.workflow.statemachine.decision_state_machine import DecisionState from cadence._internal.workflow.statemachine.decision_manager import DecisionManager from cadence._internal.workflow.statemachine.marker_state_machine import ( @@ -400,7 +402,7 @@ def test_public_get_version_dispatches_through_context(): async def test_version_marker_has_stable_id_header_and_does_not_consume_sequence(): - manager = DecisionManager(asyncio.get_event_loop()) + manager = DecisionManager(DeterministicEventLoop()) details = encode_version_marker_details(DEFAULT_VERSION) manager.version_marker_result("change", details) @@ -418,7 +420,7 @@ async def test_version_marker_has_stable_id_header_and_does_not_consume_sequence async def test_replay_preloads_python_version_marker_and_completes_its_state_machine(): - manager = DecisionManager(asyncio.get_event_loop()) + manager = DecisionManager(DeterministicEventLoop()) details = DefaultDataConverter().to_data([2]) marker_event = history.HistoryEvent( event_id=1, @@ -439,7 +441,7 @@ async def test_replay_preloads_python_version_marker_and_completes_its_state_mac async def test_replay_does_not_recreate_consumed_version_marker_in_later_batch(): - manager = DecisionManager(asyncio.get_event_loop()) + manager = DecisionManager(DeterministicEventLoop()) details = DefaultDataConverter().to_data([2]) marker_event = history.HistoryEvent( event_id=1, @@ -464,7 +466,7 @@ async def test_replay_does_not_recreate_consumed_version_marker_in_later_batch() async def test_markerless_replay_default_is_replaced_by_later_version_marker(): - manager = DecisionManager(asyncio.get_event_loop()) + manager = DecisionManager(DeterministicEventLoop()) context = Context(_info(), manager) context.set_replay_mode(True) details = DefaultDataConverter().to_data([2]) @@ -493,7 +495,7 @@ async def test_markerless_replay_default_is_replaced_by_later_version_marker(): async def test_replay_rejects_conflicting_recorded_version_markers(): - manager = DecisionManager(asyncio.get_event_loop()) + manager = DecisionManager(DeterministicEventLoop()) first_details = encode_version_marker_details(2) second_details = encode_version_marker_details(3) first_marker = history.HistoryEvent( @@ -527,7 +529,7 @@ async def test_replay_rejects_conflicting_recorded_version_markers(): async def test_version_markers_coexist_with_existing_markers_without_shifting_ids(): - manager = DecisionManager(asyncio.get_event_loop()) + manager = DecisionManager(DeterministicEventLoop()) details = DefaultDataConverter().to_data([2]) version_event = history.HistoryEvent( event_id=1, @@ -564,7 +566,7 @@ async def test_version_markers_coexist_with_existing_markers_without_shifting_id async def test_replay_ignores_foreign_version_marker_format(): - manager = DecisionManager(asyncio.get_event_loop()) + manager = DecisionManager(DeterministicEventLoop()) context = Context(_info(), manager) context.set_replay_mode(True) foreign_marker = history.HistoryEvent( @@ -587,7 +589,7 @@ async def test_replay_ignores_foreign_version_marker_format(): async def test_replay_rejects_malformed_python_version_marker_header( header_data: bytes, ): - manager = DecisionManager(asyncio.get_event_loop()) + manager = DecisionManager(DeterministicEventLoop()) malformed_marker = history.HistoryEvent( event_id=1, marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( diff --git a/tests/cadence/_internal/workflow/test_deterministic_event_loop.py b/tests/cadence/_internal/workflow/test_deterministic_event_loop.py index 4af23174..cb5ec189 100644 --- a/tests/cadence/_internal/workflow/test_deterministic_event_loop.py +++ b/tests/cadence/_internal/workflow/test_deterministic_event_loop.py @@ -1,6 +1,17 @@ -import pytest import asyncio -from cadence._internal.workflow.deterministic_event_loop import DeterministicEventLoop +import contextvars +import inspect +import logging + +import pytest +from cadence._internal.workflow.deterministic_event_loop import ( + DecisionAbortedError, + DeterministicEventLoop, + FatalDecisionError, + fail_decision, +) + +_task_var: contextvars.ContextVar[str] = contextvars.ContextVar("_task_var") async def coro_append(results: list[int], i: int): @@ -188,3 +199,270 @@ def pred2() -> bool: assert len(calls_p2) >= 2 finally: self.loop._run_forever_cleanup() + + def test_create_task_eager_start(self): + results: list[int] = [] + self.loop.create_task(coro_append(results, 1), eager_start=False) + self.loop.run_until_yield() + assert results == [1] + + coro = coro_append(results, 2) + with pytest.raises(RuntimeError, match="eager_start"): + self.loop.create_task(coro, eager_start=True) + assert inspect.getcoroutinestate(coro) == inspect.CORO_CLOSED + + def test_fail_decision_halts_loop_even_if_caught(self): + results: list[str] = [] + + async def swallow_fatal(): + try: + fail_decision(FatalDecisionError("fatal")) + except BaseException: + results.append("caught") + await asyncio.sleep(0) + results.append("resumed") + + self.loop.create_task(swallow_fatal()) + self.loop.call_soon(results.append, "other") + + with pytest.raises(FatalDecisionError, match="fatal"): + self.loop.run_until_yield() + assert results == ["caught"] + + # The loop stays failed: later runs raise without running workflow code. + with pytest.raises(FatalDecisionError, match="fatal"): + self.loop.run_until_yield() + assert results == ["caught"] + + def test_fail_decision_from_wait_condition_predicate(self): + calls_a: list[int] = [] + polls_b: list[int] = [] + waiters = [] + + def pred_a() -> bool: + calls_a.append(1) + if len(calls_a) > 1: + fail_decision(FatalDecisionError("fatal")) + return False + + def pred_b() -> bool: + polls_b.append(1) + return False + + async def register(): + # Not awaited, so settling the first waiter schedules no work. + waiters.append(self.loop.create_waiter(pred_a)) + waiters.append(self.loop.create_waiter(pred_b)) + + self.loop.create_task(register()) + with pytest.raises(FatalDecisionError, match="fatal"): + self.loop.run_until_yield() + + # The second predicate isn't polled again after the first one fails. + assert polls_b == [1] + waiters[0].exception() + + def test_close_unwinds_nested_task_groups(self): + cleanup: list[str] = [] + + async def child(level: int): + try: + await self.loop.create_future() + finally: + cleanup.append(f"child{level}") + + async def nest(level: int): + if level == 0: + await self.loop.create_future() + return + try: + # Suspended inside each body, so every __aexit__ awaits its child + # while handling the error. + async with asyncio.TaskGroup() as tg: + tg.create_task(child(level)) + await nest(level - 1) + finally: + cleanup.append(f"level{level}") + + coro = nest(3) + self.loop.create_task(coro) + self.loop.run_until_yield() + + self.loop.close() + + assert inspect.getcoroutinestate(coro) == inspect.CORO_CLOSED + assert sorted(cleanup) == [ + "child1", + "child2", + "child3", + "level1", + "level2", + "level3", + ] + + def test_close_gives_up_on_coroutine_that_never_stops_catching(self, caplog): + catches: list[int] = [] + stop = False + + async def never_stops(): + while not stop: + try: + await self.loop.create_future() + except BaseException: + catches.append(1) + + coro = never_stops() + self.loop.create_task(coro) + self.loop.run_until_yield() + + self.loop.close() + + assert len(catches) > 1 + assert inspect.getcoroutinestate(coro) == inspect.CORO_SUSPENDED + assert "could not be destroyed" in caplog.text + + # Let it exit so garbage collection doesn't hit the same loop. + stop = True + coro.close() + + @pytest.mark.parametrize( + "interact", + [ + lambda loop, coro: loop.run_until_yield(), + lambda loop, coro: loop.run_until_complete(coro), + lambda loop, coro: loop.abort_if_decision_failed(), + lambda loop, coro: loop.create_task(coro), + ], + ids=[ + "run_until_yield", + "run_until_complete", + "abort_if_decision_failed", + "create_task", + ], + ) + def test_interacting_after_close_raises_decision_aborted_error(self, interact): + self.loop.close() + coro = coro_append([], 0) + + with pytest.raises(DecisionAbortedError, match="workflow destroyed") as exc: + interact(self.loop, coro) + assert exc.value.__cause__ is None + coro.close() + + def test_close_runs_finally_in_task_context(self, caplog): + seen: list[str] = [] + + async def paused(): + token = _task_var.set("task") + try: + await self.loop.create_future() + finally: + seen.append(_task_var.get("missing")) + _task_var.reset(token) + + self.loop.create_task(paused()) + self.loop.run_until_yield() + + with caplog.at_level(logging.WARNING): + self.loop.close() + + assert seen == ["task"] + assert caplog.records == [] + + def test_close_refuses_tasks_created_by_finally(self): + ran: list[bool] = [] + refused: list[BaseException] = [] + + async def spawned(): + ran.append(True) + + async def paused(): + try: + await self.loop.create_future() + finally: + try: + asyncio.create_task(spawned()) + except FatalDecisionError as e: + refused.append(e) + + self.loop.create_task(paused()) + self.loop.run_until_yield() + + self.loop.close() + + assert ran == [] + assert len(refused) == 1 + + def test_create_task_after_caught_fatal_error_raises(self): + original = FatalDecisionError("fatal") + refused: list[BaseException] = [] + + async def keeps_going(): + try: + fail_decision(original) + except FatalDecisionError: + pass + try: + asyncio.create_task(coro_append([], 0)) + except FatalDecisionError as e: + refused.append(e) + + self.loop.create_task(keeps_going()) + with pytest.raises(FatalDecisionError) as exc_info: + self.loop.run_until_yield() + + assert exc_info.value is original + assert len(refused) == 1 + assert isinstance(refused[0], DecisionAbortedError) + assert refused[0].__cause__ is original + + def test_create_task_after_fatal_error_keeps_original_context(self): + original = FatalDecisionError("fatal") + + async def keeps_going(): + try: + fail_decision(original) + except FatalDecisionError: + pass + try: + raise KeyError("unrelated") + except KeyError: + asyncio.create_task(coro_append([], 0)) + + self.loop.create_task(keeps_going()) + with pytest.raises(FatalDecisionError) as exc_info: + self.loop.run_until_yield() + + assert exc_info.value is original + assert original.__context__ is None + + @pytest.mark.parametrize("fail_first", [True, False]) + def test_close_throws_decision_aborted_error(self, fail_first: bool): + original = FatalDecisionError("original") + seen: list[tuple[BaseException, bool]] = [] + + async def paused(): + try: + await self.loop.create_future() + except BaseException as e: + seen.append((e, asyncio.get_running_loop() is self.loop)) + raise + + async def fail(): + fail_decision(original) + + self.loop.create_task(paused()) + if fail_first: + self.loop.create_task(fail()) + with pytest.raises(FatalDecisionError): + self.loop.run_until_yield() + else: + self.loop.run_until_yield() + + self.loop.close() + + assert len(seen) == 1 + error, in_running_loop = seen[0] + assert isinstance(error, DecisionAbortedError) + assert error.__cause__ is (original if fail_first else None) + assert in_running_loop diff --git a/tests/cadence/_internal/workflow/test_marker_integration.py b/tests/cadence/_internal/workflow/test_marker_integration.py index 1f193d14..8d439553 100644 --- a/tests/cadence/_internal/workflow/test_marker_integration.py +++ b/tests/cadence/_internal/workflow/test_marker_integration.py @@ -1,4 +1,4 @@ -import asyncio +from cadence._internal.workflow.deterministic_event_loop import DeterministicEventLoop from cadence._internal.workflow.decision_events_iterator import DecisionEventsIterator from cadence._internal.workflow.statemachine.decision_manager import DecisionManager @@ -11,7 +11,7 @@ async def test_replay_marker_event_is_preloaded_before_marker_decision_exists(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) marker_attrs = decision.RecordMarkerDecisionAttributes( marker_name="SideEffect", details=Payload(data=b"recorded-value"), @@ -34,7 +34,7 @@ async def test_replay_marker_event_is_preloaded_before_marker_decision_exists(): async def test_current_decision_task_emits_record_marker_decision(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) decision_events = next(iter(DecisionEventsIterator(_current_decision_history()))) marker_attrs = decision.RecordMarkerDecisionAttributes( marker_name="MutableSideEffect", @@ -53,7 +53,7 @@ async def test_current_decision_task_emits_record_marker_decision(): async def test_multiple_replayed_marker_outputs_complete_in_decision_order(): - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) first_attrs = decision.RecordMarkerDecisionAttributes( marker_name="SideEffect", details=Payload(data=b"first"), @@ -86,7 +86,7 @@ async def test_multiple_replayed_marker_outputs_complete_in_decision_order(): async def test_version_marker_added_on_replay_is_not_nondeterministic(): # Version markers are exempt from non-determinism tracking (same as Go SDK). # A history containing a Version marker followed by a SideEffect must replay without error. - decisions = DecisionManager(asyncio.get_event_loop()) + decisions = DecisionManager(DeterministicEventLoop()) decision_events = next( iter(DecisionEventsIterator(_history_with_version_and_side_effect())) ) diff --git a/tests/cadence/_internal/workflow/test_signal_handling.py b/tests/cadence/_internal/workflow/test_signal_handling.py index 5baa145a..60125ea3 100644 --- a/tests/cadence/_internal/workflow/test_signal_handling.py +++ b/tests/cadence/_internal/workflow/test_signal_handling.py @@ -18,6 +18,10 @@ WorkflowExecutionStartedEventAttributes, WorkflowExecutionSignaledEventAttributes, ) +from cadence._internal.workflow.deterministic_event_loop import ( + FatalDecisionError, + fail_decision, +) from cadence._internal.workflow.workflow_engine import WorkflowEngine from cadence.data_converter import DefaultDataConverter from cadence.error import SignalFailure @@ -157,6 +161,19 @@ def handle_signal(self, value: str): raise ValueError(f"handler failed on: {value}") +class FatalSignalWorkflow: + @workflow.run + async def run(self): + await workflow.wait_condition(lambda: False) + + @workflow.signal(name="fatal_signal") + def handle_signal(self, value: str): + try: + fail_decision(FatalDecisionError(f"fatal: {value}")) + except BaseException: + pass + + class FailingPredicateWorkflow: """Workflow with two wait_conditions; the first predicate raises.""" @@ -649,6 +666,31 @@ def test_signal_handler_exception_fails_decision(self): assert err.value.signal_name == "bad_signal" assert "handler failed on: boom" in str(err.value) + def test_signal_handler_fatal_decision_error_fails_decision(self): + engine = make_workflow_engine(FatalSignalWorkflow) + events = [ + HistoryEvent( + event_id=1, + workflow_execution_started_event_attributes=WorkflowExecutionStartedEventAttributes( + input=Payload(data=b"[]"), + ), + ), + signal_event(2, "fatal_signal", "boom"), + HistoryEvent( + event_id=3, + decision_task_scheduled_event_attributes=DecisionTaskScheduledEventAttributes(), + ), + HistoryEvent( + event_id=4, + decision_task_started_event_attributes=DecisionTaskStartedEventAttributes( + scheduled_event_id=3, + ), + ), + ] + + with pytest.raises(FatalDecisionError, match="fatal: boom"): + engine.process_decision(events) + def test_signal_with_invalid_payload_drops_signal(self, caplog): """Malformed signal payloads are logged as warnings and dropped.""" engine = make_workflow_engine(MultiParamSignalWorkflow) diff --git a/tests/cadence/_internal/workflow/test_workflow_engine.py b/tests/cadence/_internal/workflow/test_workflow_engine.py index 46d2862a..295dce2b 100644 --- a/tests/cadence/_internal/workflow/test_workflow_engine.py +++ b/tests/cadence/_internal/workflow/test_workflow_engine.py @@ -19,6 +19,11 @@ WorkflowExecutionCompletedEventAttributes, WorkflowExecutionStartedEventAttributes, ) +from cadence._internal.workflow.deterministic_event_loop import ( + DecisionAbortedError, + FatalDecisionError, + fail_decision, +) from cadence._internal.workflow.workflow_engine import WorkflowEngine from cadence import workflow from cadence.data_converter import DefaultDataConverter @@ -137,6 +142,52 @@ async def run(self): return "unreachable" +class TaskGroupFatalWorkflow: + @workflow.run + async def run(self): + async def fail(): + fail_decision(FatalDecisionError("fatal")) + + async with asyncio.TaskGroup() as tg: + tg.create_task(fail()) + return "unreachable" + + +class CleanupOnDestroyWorkflow: + cleanup: list[object] = [] + + @workflow.run + async def run(self): + try: + await workflow.execute_activity( + "act", str, schedule_to_close_timeout=timedelta(minutes=5) + ) + finally: + self.cleanup.append(WorkflowContext.get().info().workflow_id) + try: + await workflow.execute_activity( + "cleanup", str, schedule_to_close_timeout=timedelta(minutes=5) + ) + except BaseException as e: + self.cleanup.append(type(e)) + + +class ScheduleAfterFatalWorkflow: + attempts: list[object] = [] + + @workflow.run + async def run(self): + try: + fail_decision(FatalDecisionError("fatal")) + except BaseException: + pass + try: + workflow.upsert_search_attributes({"CustomIntField": 1}) + except BaseException as e: + self.attempts.append(type(e)) + await workflow.wait_condition(lambda: False) + + class RewritingUpsertInterceptor(WorkflowInterceptor): def __init__(self, delegate: WorkflowContext) -> None: super().__init__(delegate) @@ -530,6 +581,48 @@ def factory(ctx: WorkflowContext) -> WorkflowContext: ].upsert_workflow_search_attributes_decision_attributes.search_attributes.indexed_fields assert indexed["CustomIntField"].data == b"2" + def test_fatal_decision_error_in_task_group_fails_decision(self): + workflow_engine = create_workflow_engine( + WorkflowDefinition.wrap( + TaskGroupFatalWorkflow, + WorkflowDefinitionOptions(name="task_group_fatal_workflow"), + ) + ) + + with pytest.raises(FatalDecisionError, match="fatal"): + workflow_engine.process_decision(_start_events()) + + def test_destroy_runs_workflow_finally_with_context_and_blocks_decisions(self): + CleanupOnDestroyWorkflow.cleanup = [] + workflow_engine = create_workflow_engine( + WorkflowDefinition.wrap( + CleanupOnDestroyWorkflow, + WorkflowDefinitionOptions(name="cleanup_on_destroy_workflow"), + ) + ) + workflow_engine.process_decision(_start_events()) + assert CleanupOnDestroyWorkflow.cleanup == [] + + workflow_engine.destroy() + + assert CleanupOnDestroyWorkflow.cleanup == [ + "test-workflow-id", + DecisionAbortedError, + ] + + def test_caught_fatal_decision_error_blocks_new_decisions(self): + ScheduleAfterFatalWorkflow.attempts = [] + workflow_engine = create_workflow_engine( + WorkflowDefinition.wrap( + ScheduleAfterFatalWorkflow, + WorkflowDefinitionOptions(name="schedule_after_fatal_workflow"), + ) + ) + + with pytest.raises(FatalDecisionError, match="fatal"): + workflow_engine.process_decision(_start_events()) + assert ScheduleAfterFatalWorkflow.attempts == [DecisionAbortedError] + def test_upsert_search_attributes_replay_does_not_reemit(self): workflow_engine = create_workflow_engine( WorkflowDefinition.wrap( diff --git a/tests/cadence/testing/test_workflow_environment.py b/tests/cadence/testing/test_workflow_environment.py index 9b295603..89c6387c 100644 --- a/tests/cadence/testing/test_workflow_environment.py +++ b/tests/cadence/testing/test_workflow_environment.py @@ -6,6 +6,10 @@ import pytest from cadence import workflow +from cadence._internal.workflow.deterministic_event_loop import ( + FatalDecisionError, + fail_decision, +) from cadence.testing import TestWorkflowEnvironment from cadence.worker import Registry from cadence.workflow import ( @@ -141,6 +145,31 @@ async def run(self) -> str: raise ValueError("boom") +destroyed_workflow_ids: list[str] = [] + + +@registry.workflow +class BlockedWorkflow: + @workflow.run + async def run(self) -> str: + try: + await workflow.wait_condition(lambda: False) + finally: + destroyed_workflow_ids.append(WorkflowContext.get().info().workflow_id) + return "unreachable" + + +@registry.workflow +class SwallowedFatalWorkflow: + @workflow.run + async def run(self) -> str: + try: + fail_decision(FatalDecisionError("fatal")) + except BaseException: + pass + return "unreachable" + + @registry.workflow class UpsertSearchAttributesWorkflow: @workflow.run @@ -409,6 +438,29 @@ async def test_failing_workflow(env: TestWorkflowEnvironment): env.get_workflow_result() +@pytest.mark.asyncio +async def test_close_unwinds_blocked_workflow(): + destroyed_workflow_ids.clear() + env = TestWorkflowEnvironment(registry) + await env.client.start_workflow( + "BlockedWorkflow", task_list="tl", workflow_id="blocked" + ) + assert destroyed_workflow_ids == [] + + env.close() + + assert destroyed_workflow_ids == ["blocked"] + + +@pytest.mark.asyncio +async def test_swallowed_fatal_decision_error_still_fails( + env: TestWorkflowEnvironment, +): + await env.client.start_workflow("SwallowedFatalWorkflow", task_list="tl") + assert env.is_workflow_completed() + assert isinstance(env.get_workflow_error(), FatalDecisionError) + + @pytest.mark.asyncio async def test_context_manager_closes(): with TestWorkflowEnvironment(registry) as env: diff --git a/tests/cadence/worker/test_base_task_handler.py b/tests/cadence/worker/test_base_task_handler.py index 1d3f079f..49d9f01b 100644 --- a/tests/cadence/worker/test_base_task_handler.py +++ b/tests/cadence/worker/test_base_task_handler.py @@ -12,7 +12,7 @@ def __init__(self, client, task_list: str, identity: str, **options): self._handle_task_implementation_called = False self._handle_task_failure_called = False self._last_task: str = "" - self._last_error: Exception | None = None + self._last_error: BaseException | None = None async def _handle_task_implementation(self, task: str) -> None: """Test implementation of task handling.""" @@ -21,7 +21,7 @@ async def _handle_task_implementation(self, task: str) -> None: if task == "raise_error": raise ValueError("Test error") - async def handle_task_failure(self, task: str, error: Exception) -> None: + async def handle_task_failure(self, task: str, error: BaseException) -> None: """Test implementation of task failure handling.""" self._handle_task_failure_called = True self._last_task = task @@ -85,7 +85,9 @@ class IncompleteHandler(BaseTaskHandler[str]): async def _handle_task_implementation(self, task: str) -> None: raise NotImplementedError() - async def handle_task_failure(self, task: str, error: Exception) -> None: + async def handle_task_failure( + self, task: str, error: BaseException + ) -> None: raise NotImplementedError() handler = IncompleteHandler(client, "test_task_list", "test_identity") @@ -105,7 +107,9 @@ class IntHandler(BaseTaskHandler[int]): async def _handle_task_implementation(self, task: int) -> None: pass - async def handle_task_failure(self, task: int, error: Exception) -> None: + async def handle_task_failure( + self, task: int, error: BaseException + ) -> None: pass handler = IntHandler(client, "test_task_list", "test_identity") diff --git a/tests/cadence/worker/test_metrics_propagation.py b/tests/cadence/worker/test_metrics_propagation.py index dd9cc790..827f6a5b 100644 --- a/tests/cadence/worker/test_metrics_propagation.py +++ b/tests/cadence/worker/test_metrics_propagation.py @@ -29,7 +29,7 @@ class ConcreteHandler(BaseTaskHandler[str]): async def _handle_task_implementation(self, task: str) -> None: pass - async def handle_task_failure(self, task: str, error: Exception) -> None: + async def handle_task_failure(self, task: str, error: BaseException) -> None: pass diff --git a/tests/integration_tests/nondeterminism/test_nondeterministic_workflows.py b/tests/integration_tests/nondeterminism/test_nondeterministic_workflows.py index 175b63d6..5d1c84e6 100644 --- a/tests/integration_tests/nondeterminism/test_nondeterministic_workflows.py +++ b/tests/integration_tests/nondeterminism/test_nondeterministic_workflows.py @@ -520,7 +520,6 @@ def test_nondeterministic_changes( workflow_class: Type[Any], expected: NonDeterminismError | None, ) -> None: - # pytest.raises is awkward with ExceptionGroups, use try/except instead try: decisions = replay_with_history(helper, workflow_class) if expected is not None: @@ -538,15 +537,6 @@ def test_nondeterministic_changes( assert e.args[1:] == expected.args[1:] else: raise e - except ExceptionGroup as e: - # Only look at the first NonDeterminismError - if expected is not None: - subgroup = e.subgroup(NonDeterminismError) - assert subgroup is not None - errors = subgroup.exceptions - assert errors[0].args[1:] == expected.args[1:] - else: - raise e def replay_with_history(helper: CadenceHelper, clazz: Type[Any]) -> DecisionResult: @@ -565,6 +555,9 @@ def replay_with_history(helper: CadenceHelper, clazz: Type[Any]) -> DecisionResu ), workflow_def, ) - result = engine.process_decision(list(history.events)) - assert engine.is_done() - return result + try: + result = engine.process_decision(list(history.events)) + assert engine.is_done() + return result + finally: + engine.destroy() diff --git a/tests/integration_tests/workflow/test_fatal_decision.py b/tests/integration_tests/workflow/test_fatal_decision.py new file mode 100644 index 00000000..cd41dc73 --- /dev/null +++ b/tests/integration_tests/workflow/test_fatal_decision.py @@ -0,0 +1,102 @@ +import asyncio +from datetime import timedelta + +import pytest + +from cadence import Registry, workflow +from cadence.client import Client +from cadence.api.v1.common_pb2 import WorkflowExecution +from cadence.api.v1.history_pb2 import HistoryEvent +from cadence.api.v1.service_workflow_pb2 import ( + GetWorkflowExecutionHistoryRequest, + TerminateWorkflowExecutionRequest, +) +from tests.integration_tests.helper import CadenceHelper, DOMAIN_NAME + +registry = Registry() + + +def _incompatible_versions() -> None: + # The first call records version 1, so the second call resolves to 1, which is + # outside its supported range and raises a FatalDecisionError. + workflow.get_version("change", 1, 1) + workflow.get_version("change", 2, 2) + + +@registry.workflow() +class IncompatibleVersionWorkflow: + @workflow.run + async def run(self) -> str: + _incompatible_versions() + return "completed" + + +@registry.workflow() +class SwallowedFatalDecisionWorkflow: + @workflow.run + async def run(self) -> str: + try: + _incompatible_versions() + except BaseException: + pass + return "swallowed" + + +async def _events_until_decision_task_failed( + client: Client, execution: WorkflowExecution +) -> list[HistoryEvent]: + """Long poll the history until the first DecisionTaskFailed event.""" + events: list[HistoryEvent] = [] + next_page_token = b"" + async with asyncio.timeout(10): + while True: + response = await client.workflow_stub.GetWorkflowExecutionHistory( + GetWorkflowExecutionHistoryRequest( + domain=DOMAIN_NAME, + workflow_execution=execution, + next_page_token=next_page_token, + wait_for_new_event=True, + skip_archival=True, + ) + ) + for event in response.history.events: + events.append(event) + if event.HasField("decision_task_failed_event_attributes"): + return events + if not response.next_page_token: + raise AssertionError("Workflow closed without failing a decision task") + next_page_token = response.next_page_token + + +@pytest.mark.parametrize( + "workflow_type", + ["IncompatibleVersionWorkflow", "SwallowedFatalDecisionWorkflow"], +) +async def test_fatal_decision_error_fails_decision_task( + helper: CadenceHelper, workflow_type: str +) -> None: + async with helper.worker(registry) as worker: + execution = await worker.client.start_workflow( + workflow_type, + task_list=worker.task_list, + execution_start_to_close_timeout=timedelta(seconds=30), + ) + + events = await _events_until_decision_task_failed(worker.client, execution) + + failed = events[-1].decision_task_failed_event_attributes + assert b"outside the supported range" in failed.failure.details + assert not any( + event.HasField("decision_task_completed_event_attributes") + or event.HasField("marker_recorded_event_attributes") + or event.HasField("workflow_execution_completed_event_attributes") + for event in events + ) + + await worker.client.workflow_stub.TerminateWorkflowExecution( + TerminateWorkflowExecutionRequest( + domain=DOMAIN_NAME, + workflow_execution=execution, + reason="test complete", + ) + )