From 52ac5aad9b8711b977f1c3e0ce03d8253b3fb70b Mon Sep 17 00:00:00 2001 From: Nathanael Mortensen Date: Mon, 19 Oct 2026 09:33:53 -0700 Subject: [PATCH] feat: Consistently Handle FatalDecisionError WorkflowCode can catch FatalDecisionError and often does so accidentally via `except Exception`. Consistently fail the decision: - Store the FatalDecisionError on DeterministicEventLoop and stop processing coroutines once it has failed. Reraise it to the WorkflowEngine for reporting. - Change FatalDecisionError to extend BaseException rather than Exception. This is a more correct base class that users are unlikely to accidentally catch. We want to collect and report only the first FatalDecisionError as WorkflowCode might incidentally cause additional errors once that has happened and they're likely red herrings. Once we've completed a decision, successfully or otherwise, we currently discard the WorkflowEngine and let the garbage collector clean it up. This is problematic for a number of reasons: - Python logs warnings about these unawaited coroutines. - As part of garbage collection Python `cancel`s them, which executes the user code outside of the WorkflowContext and event loop. - We want to consistently run `finally` blocks, even when destroying an incomplete Workflow. This sounds strange, but: - In Go we use goexit which executes `defer` blocks. - In Java we throw DestroyWorkflowThreadError which executes `finally` blocks. As we provide the event loop we're responsible for running their coroutines to completion. They might have an `await` within a `finally` block, especially if using an async context manager, which means we need to repeatedly run and interrupt it. We don't know how many times their coroutine will do this, or if it's doing it infinitely, so we try at most 100 times before giving up on a coroutine and leaving it for the garbage collector. Destroy the WorkflowEngine after each DecisionTask. --- cadence/_internal/workflow/context.py | 11 +- .../workflow/deterministic_event_loop.py | 209 +++++++++++-- .../workflow/statemachine/decision_manager.py | 35 ++- .../workflow/statemachine/nondeterminism.py | 15 +- cadence/_internal/workflow/versioning.py | 13 +- cadence/_internal/workflow/workflow_engine.py | 10 +- cadence/testing/_workflow_environment.py | 41 ++- cadence/worker/_base_task_handler.py | 5 +- cadence/worker/_decision_task_handler.py | 23 +- .../statemachine/test_decision_manager.py | 82 ++--- .../workflow/test_context_versioning.py | 22 +- .../workflow/test_deterministic_event_loop.py | 282 +++++++++++++++++- .../workflow/test_marker_integration.py | 10 +- .../workflow/test_signal_handling.py | 42 +++ .../workflow/test_workflow_engine.py | 93 ++++++ .../testing/test_workflow_environment.py | 52 ++++ .../cadence/worker/test_base_task_handler.py | 12 +- .../worker/test_metrics_propagation.py | 2 +- .../test_nondeterministic_workflows.py | 19 +- .../workflow/test_fatal_decision.py | 102 +++++++ 20 files changed, 945 insertions(+), 135 deletions(-) create mode 100644 tests/integration_tests/workflow/test_fatal_decision.py 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", + ) + )