diff --git a/cadence/_internal/activity/_activity_executor.py b/cadence/_internal/activity/_activity_executor.py index e213926..d522f36 100644 --- a/cadence/_internal/activity/_activity_executor.py +++ b/cadence/_internal/activity/_activity_executor.py @@ -3,16 +3,16 @@ from datetime import datetime, timedelta, timezone from logging import getLogger import time -from traceback import format_exception from typing import Any, Callable, Optional, Sequence, Union, cast from google.protobuf.duration import to_timedelta from google.protobuf.timestamp import to_datetime from cadence._internal.activity._context import _Context, _SyncContext from cadence._internal.context import header_to_dict +from cadence._internal.failure import failure_from_exception from cadence._internal.activity._definition import BaseDefinition, ExecutionStrategy from cadence._internal.activity._heartbeat import _HeartbeatSender from cadence.activity import ActivityInfo, ActivityDefinition -from cadence.api.v1.common_pb2 import Failure, Payload +from cadence.api.v1.common_pb2 import Payload from cadence.api.v1.service_worker_pb2 import ( PollForActivityTaskResponse, RespondActivityTaskCanceledRequest, @@ -41,7 +41,7 @@ TAG_TASK_LIST, TAG_WORKFLOW_TYPE, ) -from cadence.error import ActivityCancelledError +from cadence.error import ActivityCancelledError, ApplicationError, FailureCategory _logger = getLogger(__name__) @@ -118,7 +118,11 @@ async def execute(self, task: PollForActivityTaskResponse) -> None: ) if error is not None: emitter.counter(ACTIVITY_EXECUTION_FAILED_COUNTER) - _logger.error("Activity failed", exc_info=error) + if _is_failure_category_poll(error): + # An expected failure, so it isn't logged as an activity error + _logger.debug("Activity failed", exc_info=error) + else: + _logger.error("Activity failed", exc_info=error) await self._report_failure(task, error, emitter, e2e_latency) else: await self._report_success(task, result, emitter, e2e_latency) @@ -172,7 +176,9 @@ async def _report_failure( await self._client.worker_stub.RespondActivityTaskFailed( RespondActivityTaskFailedRequest( task_token=task.task_token, - failure=_to_failure(error), + failure=failure_from_exception( + error, self._data_converter, include_options=True + ), identity=self._identity, ) ) @@ -261,10 +267,8 @@ def _create_info(self, task: PollForActivityTaskResponse) -> ActivityInfo: ) -def _to_failure(exception: Exception) -> Failure: - stacktrace = "".join(format_exception(exception)) - - return Failure( - reason=type(exception).__name__, - details=stacktrace.encode(), +def _is_failure_category_poll(error: Exception) -> bool: + return ( + isinstance(error, ApplicationError) + and error.failure_category is FailureCategory.POLL ) diff --git a/cadence/_internal/failure.py b/cadence/_internal/failure.py new file mode 100644 index 0000000..73b31fd --- /dev/null +++ b/cadence/_internal/failure.py @@ -0,0 +1,288 @@ +"""Conversion between Python exceptions and Cadence ``Failure`` messages. + +Wire format: + +- An :class:`ApplicationError` is recorded with its own reason, and its details + encoded by the data converter. This matches ``CustomError`` in the Go client. +- Any other exception is recorded with its class name as the reason, and a + UTF-8 message and stack trace as the details. +""" + +import logging +from datetime import timedelta +from typing import Any +from traceback import format_exception + +from cadence._internal.workflow.retry_policy import _set_duration_field +from cadence.api.v1 import common_pb2, decision, history, workflow_pb2 +from cadence.api.v1.common_pb2 import Failure, FailureOptions, Payload +from cadence.data_converter import DataConverter +from cadence.error import ( + ActivityError, + ApplicationError, + CadenceTimeoutError, + WorkflowCanceledError, + ChildWorkflowError, + FailureCategory, + FailureError, + SignalExternalWorkflowError, + StartChildWorkflowExecutionError, + TerminatedError, + TimeoutType, +) + +_logger = logging.getLogger(__name__) + + +def failure_from_exception( + exc: BaseException, data_converter: DataConverter, *, include_options: bool +) -> Failure: + """Convert an exception raised by activity or workflow code into a ``Failure``. + + ``include_options`` attaches an ``ApplicationError``'s failure category and + next retry delay. They are only supported for activity failures. + """ + if not isinstance(exc, ApplicationError): + return Failure(reason=type(exc).__name__, details=_describe(exc)) + + failure = Failure(reason=exc.reason, details=_encode_details(exc, data_converter)) + if include_options: + options = _failure_options(exc) + if options is not None: + failure.options.CopyFrom(options) + return failure + + +def application_error_from_failure( + failure: Failure, data_converter: DataConverter +) -> ApplicationError: + failure_category: FailureCategory | None = None + next_retry_delay: timedelta | None = None + if failure.HasField("options"): + if failure.options.failure_category != common_pb2.FAILURE_CATEGORY_INVALID: + failure_category = FailureCategory(failure.options.failure_category) + if failure.options.HasField("next_retry_interval"): + next_retry_delay = failure.options.next_retry_interval.ToTimedelta() + return ApplicationError._from_payload( + failure.reason, + Payload(data=failure.details), + data_converter, + failure_category=failure_category, + next_retry_delay=next_retry_delay, + ) + + +def activity_failed_error( + event: history.ActivityTaskFailedEventAttributes, + request: decision.ScheduleActivityTaskDecisionAttributes, + data_converter: DataConverter, +) -> ActivityError: + cause = application_error_from_failure(event.failure, data_converter) + return _activity_error( + "activity failed", + request.activity_type.name, + request.activity_id, + cause, + ) + + +def activity_timed_out_error( + event: history.ActivityTaskTimedOutEventAttributes, + request: decision.ScheduleActivityTaskDecisionAttributes, + data_converter: DataConverter, +) -> ActivityError: + # The last failure is deliberately not exposed, matching the Java client. + return _activity_error( + "activity timed out", + request.activity_type.name, + request.activity_id, + CadenceTimeoutError._from_payload( + TimeoutType(event.timeout_type), event.details, data_converter + ), + ) + + +def child_workflow_failed_error( + event: history.ChildWorkflowExecutionFailedEventAttributes, + data_converter: DataConverter, +) -> ChildWorkflowError: + return _child_workflow_error( + "child workflow failed", + event.workflow_type.name, + event.workflow_execution.workflow_id, + application_error_from_failure(event.failure, data_converter), + ) + + +def child_workflow_timed_out_error( + event: history.ChildWorkflowExecutionTimedOutEventAttributes, +) -> ChildWorkflowError: + return _child_workflow_error( + "child workflow timed out", + event.workflow_type.name, + event.workflow_execution.workflow_id, + CadenceTimeoutError(TimeoutType(event.timeout_type)), + ) + + +def child_workflow_canceled_error( + event: history.ChildWorkflowExecutionCanceledEventAttributes, + data_converter: DataConverter, +) -> ChildWorkflowError: + return _child_workflow_error( + "child workflow canceled", + event.workflow_type.name, + event.workflow_execution.workflow_id, + WorkflowCanceledError._from_payload(event.details, data_converter), + ) + + +def child_workflow_terminated_error( + event: history.ChildWorkflowExecutionTerminatedEventAttributes, +) -> ChildWorkflowError: + return _child_workflow_error( + "child workflow terminated", + event.workflow_type.name, + event.workflow_execution.workflow_id, + TerminatedError(), + ) + + +def start_child_workflow_failed_error( + event: history.StartChildWorkflowExecutionFailedEventAttributes, + request: decision.StartChildWorkflowExecutionDecisionAttributes, +) -> StartChildWorkflowExecutionError: + cause = _enum_name( + workflow_pb2.ChildWorkflowExecutionFailedCause, + event.cause, + "CHILD_WORKFLOW_EXECUTION_FAILED_CAUSE_", + ) + return StartChildWorkflowExecutionError( + f"start child workflow failed (workflow_type={request.workflow_type.name}, " + f"workflow_id={event.workflow_id}): {cause}", + failure_cause=event.cause, + workflow_id=event.workflow_id, + workflow_type=request.workflow_type.name, + ) + + +def signal_external_workflow_error( + event: history.SignalExternalWorkflowExecutionFailedEventAttributes, + request: decision.SignalExternalWorkflowExecutionDecisionAttributes, +) -> SignalExternalWorkflowError: + cause = _enum_name( + workflow_pb2.SignalExternalWorkflowExecutionFailedCause, + event.cause, + "SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_", + ) + workflow_id = request.workflow_execution.workflow_id + return SignalExternalWorkflowError( + f"signal external workflow failed (signal_name={request.signal_name}, " + f"workflow_id={workflow_id}): {cause}", + failure_cause=event.cause, + workflow_id=workflow_id, + signal_name=request.signal_name, + ) + + +def activity_error_from_exception( + exc: Exception, + activity_type: str, + activity_id: str, + data_converter: DataConverter, +) -> ActivityError: + """Build the error a workflow would see if an activity raised ``exc``. + + For running activities in-process, such as in the test environment. + """ + failure = failure_from_exception(exc, data_converter, include_options=True) + cause = application_error_from_failure(failure, data_converter) + return _activity_error("activity failed", activity_type, activity_id, cause) + + +def child_workflow_error_from_exception( + exc: Exception, + workflow_type: str, + workflow_id: str, + data_converter: DataConverter, +) -> ChildWorkflowError: + """Build the error a parent workflow would see if a child workflow raised ``exc``. + + For running child workflows in-process, such as in the test environment. + """ + failure = failure_from_exception(exc, data_converter, include_options=False) + cause = application_error_from_failure(failure, data_converter) + return _child_workflow_error( + "child workflow failed", workflow_type, workflow_id, cause + ) + + +def _child_workflow_error( + summary: str, + workflow_type: str, + workflow_id: str, + cause: FailureError, +) -> ChildWorkflowError: + error = ChildWorkflowError( + f"{summary} (workflow_type={workflow_type}, workflow_id={workflow_id}): " + f"{cause}", + workflow_id=workflow_id, + workflow_type=workflow_type, + ) + # These errors are set on futures rather than raised, so chain them explicitly + error.__cause__ = cause + return error + + +def _activity_error( + summary: str, + activity_type: str, + activity_id: str, + cause: ApplicationError | CadenceTimeoutError, +) -> ActivityError: + error = ActivityError( + f"{summary} (activity_type={activity_type}, activity_id={activity_id}): " + f"{cause}", + activity_id=activity_id, + activity_type=activity_type, + ) + # These errors are set on futures rather than raised, so chain them explicitly + error.__cause__ = cause + return error + + +def _enum_name(enum_type: Any, value: int, prefix: str) -> str: + try: + name: str = enum_type.Name(value) + except ValueError: + return str(value) + return name.removeprefix(prefix) + + +def _describe(exc: BaseException) -> bytes: + stacktrace = "".join(format_exception(exc)) + return f"message: {exc}\nstacktrace: {stacktrace}".encode("utf-8") + + +def _encode_details(error: ApplicationError, data_converter: DataConverter) -> bytes: + # A failure received from Cadence is passed through unchanged + if error._payload is not None: + return error._payload.data + if not error.has_details(): + return b"" + try: + return data_converter.to_data(error.details()).data + except Exception as e: + _logger.warning("Failed to encode ApplicationError details", exc_info=True) + return f"failed to encode ApplicationError details: {e!r}".encode("utf-8") + + +def _failure_options(error: ApplicationError) -> FailureOptions | None: + if error.failure_category is None and error.next_retry_delay is None: + return None + options = FailureOptions() + if error.failure_category is not None: + options.failure_category = error.failure_category.value + if error.next_retry_delay is not None: + _set_duration_field(options.next_retry_interval, error.next_retry_delay) + return options diff --git a/cadence/_internal/workflow/statemachine/activity_state_machine.py b/cadence/_internal/workflow/statemachine/activity_state_machine.py index 684b61a..2fd0d11 100644 --- a/cadence/_internal/workflow/statemachine/activity_state_machine.py +++ b/cadence/_internal/workflow/statemachine/activity_state_machine.py @@ -9,10 +9,10 @@ from cadence._internal.workflow.statemachine.nondeterminism import ( record_immediate_cancel, ) +from cadence._internal.failure import activity_failed_error, activity_timed_out_error from cadence.api.v1 import decision, history from cadence.api.v1.common_pb2 import Payload -from cadence.api.v1.workflow_pb2 import TimeoutType -from cadence.error import ActivityFailure +from cadence.data_converter import DataConverter activity_events = EventDispatcher("scheduled_event_id") @@ -25,10 +25,12 @@ def __init__( self, request: decision.ScheduleActivityTaskDecisionAttributes, completed: DecisionFuture[Payload], + data_converter: DataConverter, ) -> None: super().__init__() self.request = request self.completed = completed + self._data_converter = data_converter def get_id(self) -> DecisionId: return DecisionId(DecisionType.ACTIVITY, self.request.activity_id) @@ -87,12 +89,9 @@ def handle_completed( @activity_events.event() def handle_failed(self, event: history.ActivityTaskFailedEventAttributes) -> None: self._transition(DecisionState.COMPLETED) - failure_details = ( - event.failure.details.decode("utf-8", errors="replace") or None - ) self._resolve( self.completed, - exc=ActivityFailure(event.failure.reason, failure_details), + exc=activity_failed_error(event, self.request, self._data_converter), ) @activity_events.event() @@ -100,16 +99,9 @@ def handle_timeout( self, event: history.ActivityTaskTimedOutEventAttributes ) -> None: self._transition(DecisionState.COMPLETED) - failure_details = ( - event.last_failure.details.decode("utf-8", errors="replace") or None - ) self._resolve( self.completed, - exc=ActivityFailure( - TimeoutType.Name(event.timeout_type), - failure_details=failure_details, - heartbeat_details=event.details, - ), + exc=activity_timed_out_error(event, self.request, self._data_converter), ) @activity_events.event() diff --git a/cadence/_internal/workflow/statemachine/child_workflow_execution_state_machine.py b/cadence/_internal/workflow/statemachine/child_workflow_execution_state_machine.py index 2ba3a76..6e696c6 100644 --- a/cadence/_internal/workflow/statemachine/child_workflow_execution_state_machine.py +++ b/cadence/_internal/workflow/statemachine/child_workflow_execution_state_machine.py @@ -11,15 +11,16 @@ from cadence._internal.workflow.statemachine.nondeterminism import ( record_immediate_cancel, ) +from cadence._internal.failure import ( + child_workflow_canceled_error, + child_workflow_failed_error, + child_workflow_terminated_error, + child_workflow_timed_out_error, + start_child_workflow_failed_error, +) from cadence.api.v1 import decision, history from cadence.api.v1.common_pb2 import Payload, WorkflowExecution -from cadence.error import ( - ChildWorkflowExecutionCanceled, - ChildWorkflowExecutionFailed, - ChildWorkflowExecutionTerminated, - ChildWorkflowExecutionTimedOut, - StartChildWorkflowExecutionFailed, -) +from cadence.data_converter import DataConverter child_workflow_events = EventDispatcher("initiated_event_id") @@ -37,11 +38,13 @@ def __init__( request: decision.StartChildWorkflowExecutionDecisionAttributes, execution: DecisionFuture[WorkflowExecution], result: DecisionFuture[Payload], + data_converter: DataConverter, ) -> None: super().__init__() self.request = request self.execution = execution self.result = result + self._data_converter = data_converter self._run_id = None def get_id(self) -> DecisionId: @@ -106,11 +109,7 @@ def handle_initiation_failed( self, event: history.StartChildWorkflowExecutionFailedEventAttributes ) -> None: self._transition(DecisionState.COMPLETED) - exc = StartChildWorkflowExecutionFailed( - f"start child failed: {event.cause}", - cause=event.cause, - workflow_id=event.workflow_id, - ) + exc = start_child_workflow_failed_error(event, self.request) self._resolve(self.execution, exc=exc) self._resolve(self.result, exc=exc) @@ -139,43 +138,44 @@ def handle_failed( self._transition(DecisionState.COMPLETED) self._resolve( self.result, - exc=ChildWorkflowExecutionFailed( - event.failure.reason, - failure=event.failure, - ), + exc=child_workflow_failed_error(event, self._data_converter), ) @child_workflow_events.event() def handle_canceled( self, event: history.ChildWorkflowExecutionCanceledEventAttributes ) -> None: - self._transition(DecisionState.COMPLETED) - self._resolve( - self.result, - exc=ChildWorkflowExecutionCanceled( - "child workflow canceled", details=event.details - ), + cancel_requested = self.state in ( + DecisionState.CANCELED_AFTER_RECORDED, + DecisionState.CANCELED_AFTER_STARTED, + DecisionState.CANCELLATION_RECORDED, ) + self._transition(DecisionState.COMPLETED) + if self.result.done(): + return + if cancel_requested: + self.result.force_cancel(event.details.data.decode()) + else: + # Canceled by something other than this workflow, so it's a failure + # of the child rather than cancellation of the awaiting coroutine. + self._resolve( + self.result, + exc=child_workflow_canceled_error(event, self._data_converter), + ) @child_workflow_events.event() def handle_timed_out( self, event: history.ChildWorkflowExecutionTimedOutEventAttributes ) -> None: self._transition(DecisionState.COMPLETED) - self._resolve( - self.result, - exc=ChildWorkflowExecutionTimedOut( - f"child workflow timed out: {event.timeout_type}", - timeout_type=int(event.timeout_type), - ), - ) + self._resolve(self.result, exc=child_workflow_timed_out_error(event)) @child_workflow_events.event() def handle_terminated( - self, _: history.ChildWorkflowExecutionTerminatedEventAttributes + self, event: history.ChildWorkflowExecutionTerminatedEventAttributes ) -> None: self._transition(DecisionState.COMPLETED) - self._resolve(self.result, exc=ChildWorkflowExecutionTerminated()) + self._resolve(self.result, exc=child_workflow_terminated_error(event)) @child_workflow_events.event( "workflow_execution.workflow_id", event_id_is_alias=True diff --git a/cadence/_internal/workflow/statemachine/decision_manager.py b/cadence/_internal/workflow/statemachine/decision_manager.py index 5c8b6d1..053430d 100644 --- a/cadence/_internal/workflow/statemachine/decision_manager.py +++ b/cadence/_internal/workflow/statemachine/decision_manager.py @@ -61,6 +61,7 @@ ) from cadence.api.v1 import decision, history from cadence.api.v1.common_pb2 import Payload, WorkflowExecution +from cadence.data_converter import DataConverter, DefaultDataConverter logger = logging.getLogger(__name__) @@ -125,8 +126,15 @@ class DecisionManager: } ) - def __init__(self, event_loop: asyncio.AbstractEventLoop): + def __init__( + self, + event_loop: asyncio.AbstractEventLoop, + data_converter: DataConverter | None = None, + ): self._event_loop = event_loop + self._data_converter = ( + data_converter if data_converter is not None else DefaultDataConverter() + ) self._id_counter = 0 self._determinism_tracker = DeterminismTracker() self._replaying = False @@ -148,7 +156,7 @@ def schedule_activity( self._determinism_tracker.validate_action(attrs) decision_id = DecisionId(DecisionType.ACTIVITY, attrs.activity_id) future: DecisionFuture[Payload] = self._create_future(decision_id) - machine = ActivityStateMachine(attrs, future) + machine = ActivityStateMachine(attrs, future, self._data_converter) self._add_state_machine(machine) return future @@ -184,7 +192,9 @@ def schedule_child_workflow( execution: DecisionFuture[WorkflowExecution] = self._create_future(decision_id) execution.add_done_callback(_consume_future_exception) result: DecisionFuture[Payload] = self._create_future(decision_id) - machine = ChildWorkflowExecutionStateMachine(attrs, execution, result) + machine = ChildWorkflowExecutionStateMachine( + attrs, execution, result, self._data_converter + ) self._add_state_machine(machine) return execution, result diff --git a/cadence/_internal/workflow/statemachine/signal_external_workflow_state_machine.py b/cadence/_internal/workflow/statemachine/signal_external_workflow_state_machine.py index 1ebee1e..9073574 100644 --- a/cadence/_internal/workflow/statemachine/signal_external_workflow_state_machine.py +++ b/cadence/_internal/workflow/statemachine/signal_external_workflow_state_machine.py @@ -8,8 +8,8 @@ DecisionType, ) from cadence._internal.workflow.statemachine.event_dispatcher import EventDispatcher +from cadence._internal.failure import signal_external_workflow_error from cadence.api.v1 import decision, history -from cadence.error import SignalExternalWorkflowFailed signal_external_events = EventDispatcher("initiated_event_id") @@ -68,8 +68,5 @@ def handle_failed( self._transition(DecisionState.COMPLETED) self._resolve( self.completed, - exc=SignalExternalWorkflowFailed( - f"signal external workflow failed: {event.cause}", - cause=event.cause, - ), + exc=signal_external_workflow_error(event, self.request), ) diff --git a/cadence/_internal/workflow/workflow_engine.py b/cadence/_internal/workflow/workflow_engine.py index 1b5d5ef..bec0eb4 100644 --- a/cadence/_internal/workflow/workflow_engine.py +++ b/cadence/_internal/workflow/workflow_engine.py @@ -1,11 +1,11 @@ import logging -import traceback from asyncio import CancelledError, InvalidStateError from dataclasses import dataclass from functools import singledispatchmethod from typing import List, Mapping, Optional, Sequence from cadence._internal.context import extract_headers, set_header_from_dict +from cadence._internal.failure import failure_from_exception from cadence._internal.workflow.context import Context, _activate_workflow_context from cadence._internal.workflow.decision_events_iterator import DecisionEventsIterator from cadence._internal.workflow.deterministic_event_loop import ( @@ -21,7 +21,7 @@ CompleteWorkflowExecutionDecisionAttributes, ContinueAsNewWorkflowExecutionDecisionAttributes, ) -from cadence.api.v1.common_pb2 import Failure, Payload, WorkflowType +from cadence.api.v1.common_pb2 import Payload, WorkflowType from cadence.api.v1.history_pb2 import ( HistoryEvent, WorkflowExecutionCancelRequestedEventAttributes, @@ -62,7 +62,7 @@ def __init__( workflow_interceptor_factory: WorkflowInterceptorFactory | None = None, ): self._event_loop = DeterministicEventLoop() - self._decision_manager = DecisionManager(self._event_loop) + self._decision_manager = DecisionManager(self._event_loop, info.data_converter) self._data_converter = info.data_converter self._workflow_definition = workflow_definition self._workflow_instance = WorkflowInstance( @@ -203,9 +203,12 @@ def _process_decision_events( # Signal handler failures fail the decision task, not the workflow. if ( - signal_failure := self._workflow_instance.get_signal_failure() + signal_error := self._workflow_instance.get_signal_handler_error() ) is not None: - raise signal_failure + signal_name, exc = signal_error + raise FatalDecisionError( + f"signal handler {signal_name!r} failed: {exc!r}" + ) from exc if decision := self._maybe_complete_workflow(): self._decision_manager.complete_workflow(decision) @@ -260,7 +263,9 @@ def _maybe_complete_workflow(self) -> Optional[Decision]: except ExceptionGroup as e: if e.subgroup((InvalidStateError, FatalDecisionError)): raise - failure = _failure_from_exception(e) + failure = failure_from_exception( + e, self._data_converter, include_options=False + ) return Decision( fail_workflow_execution_decision_attributes=FailWorkflowExecutionDecisionAttributes( @@ -269,7 +274,9 @@ def _maybe_complete_workflow(self) -> Optional[Decision]: ) except Exception as e: - failure = _failure_from_exception(e) + failure = failure_from_exception( + e, self._data_converter, include_options=False + ) return Decision( fail_workflow_execution_decision_attributes=FailWorkflowExecutionDecisionAttributes( failure=failure @@ -341,14 +348,3 @@ def _outcome_from_decision(decision: Decision) -> Optional[str]: if attr == "continue_as_new_workflow_execution_decision_attributes": return "continue_as_new" return None - - -def _failure_from_exception(e: Exception) -> Failure: - stacktrace = "".join(traceback.format_exception(e)) - - details = f"message: {str(e)}\nstacktrace: {stacktrace}" - - return Failure( - reason=type(e).__name__, - details=details.encode("utf-8"), - ) diff --git a/cadence/_internal/workflow/workflow_instance.py b/cadence/_internal/workflow/workflow_instance.py index f73fcfe..fa73e82 100644 --- a/cadence/_internal/workflow/workflow_instance.py +++ b/cadence/_internal/workflow/workflow_instance.py @@ -6,7 +6,6 @@ from cadence._internal.workflow.deterministic_event_loop import ( DeterministicEventLoop, ) -from cadence.error import SignalFailure from cadence.query import QueryDefinition from cadence.signal import SignalDefinition from cadence.workflow import WorkflowCancellationInfo, WorkflowDefinition @@ -34,10 +33,11 @@ def __init__( # Strong references to in-flight async signal handler tasks. self._signal_tasks: set[Task[Any]] = set() # Fail the decision task if a signal handler raises an exception. - self._signal_failure: Optional[SignalFailure] = None + self._signal_handler_error: Optional[tuple[str, Exception]] = None - def get_signal_failure(self) -> Optional[SignalFailure]: - return self._signal_failure + def get_signal_handler_error(self) -> Optional[tuple[str, Exception]]: + """Return the name of the first signal whose handler raised, and its exception.""" + return self._signal_handler_error def start(self, args: list[Any]) -> None: if self._task is None: @@ -119,5 +119,5 @@ def _on_signal_task_done(self, task: Task[Any], signal_name: str) -> None: if task.cancelled(): return exc = task.exception() - if isinstance(exc, Exception) and self._signal_failure is None: - self._signal_failure = SignalFailure(str(exc) or None, signal_name) + if isinstance(exc, Exception) and self._signal_handler_error is None: + self._signal_handler_error = (signal_name, exc) diff --git a/cadence/error.py b/cadence/error.py index f823ebd..cbfed45 100644 --- a/cadence/error.py +++ b/cadence/error.py @@ -1,12 +1,42 @@ from datetime import timedelta -from typing import Any +from enum import IntEnum +from typing import TYPE_CHECKING, Any import grpc +from cadence.api.v1 import common_pb2, workflow_pb2 from cadence.api.v1.common_pb2 import Payload +if TYPE_CHECKING: + from cadence.data_converter import DataConverter -class ContinueAsNewError(Exception): + +class FailureCategory(IntEnum): + """Classifies an activity failure, independently of the activity's RetryPolicy. + + The category can influence retry and logging behavior. It is currently only + supported for activity failures. + """ + + POLL = common_pb2.FAILURE_CATEGORY_POLL + """An expected failure, such as a polling activity with nothing to do yet. + The worker does not log it as an activity error.""" + STANDARD = common_pb2.FAILURE_CATEGORY_STANDARD + FATAL = common_pb2.FAILURE_CATEGORY_FATAL + + +class TimeoutType(IntEnum): + START_TO_CLOSE = workflow_pb2.TIMEOUT_TYPE_START_TO_CLOSE + SCHEDULE_TO_START = workflow_pb2.TIMEOUT_TYPE_SCHEDULE_TO_START + SCHEDULE_TO_CLOSE = workflow_pb2.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE + HEARTBEAT = workflow_pb2.TIMEOUT_TYPE_HEARTBEAT + + +class CadenceError(Exception): + """Base class for all errors raised by the Cadence client.""" + + +class ContinueAsNewError(CadenceError): def __init__( self, *args: Any, @@ -25,92 +55,273 @@ def __init__( self.headers = headers -class ActivityFailure(Exception): - """An activity failed or timed out. +class ActivityCancelledError(CadenceError): + def __init__(self, *details: Any) -> None: + super().__init__("Activity cancelled") + self.details = list(details) - ``failure_details`` contains UTF-8 diagnostics from a Cadence ``Failure``. - ``heartbeat_details`` is the raw, data-converter-encoded payload from the - activity's last heartbeat. Decode it with the same converter configured on - the client, for example:: - values = data_converter.from_data( - exc.heartbeat_details, [ExpectedType] - ) +class FailureError(CadenceError): + """Base class for failures that are recorded in Cadence and raised into workflow code.""" + + def __init__(self, message: str) -> None: + super().__init__(message) + self.message = message + + +class ApplicationError(FailureError): + """A failure with a reason and optional details, raised from activity or workflow code. + + This is the counterpart of ``CustomError`` in the Go client. ``reason`` is + recorded as the failure reason, which is what a RetryPolicy's + ``non_retryable_error_reasons`` is matched against. ``details`` are encoded + with the client's data converter, so workflows written with any Cadence + client can decode them. + + ``failure_category`` and ``next_retry_delay`` are only supported for + activity failures: + + - ``failure_category`` can influence retry and logging behavior + independently of the RetryPolicy. Failures with + :attr:`FailureCategory.POLL` are not logged as activity errors. + - ``next_retry_delay`` overrides the delay until the next attempt of the + activity. It does not influence whether the activity is retried, which is + still determined by the RetryPolicy and the failure category. The delay + is rounded up to whole seconds. + + When an activity or child workflow fails, workflow code receives an + :class:`ActivityError` or :class:`ChildWorkflowError` whose + ``cause`` is an ``ApplicationError``. Use :meth:`details` to decode its + details. """ def __init__( self, - message: str, - failure_details: str | None = None, - heartbeat_details: Payload | None = None, + reason: str, + *details: Any, + failure_category: FailureCategory | None = None, + next_retry_delay: timedelta | None = None, ) -> None: - super().__init__( - message if not failure_details else f"{message}\n{failure_details}" + if failure_category is not None and not isinstance( + failure_category, FailureCategory + ): + raise TypeError( + f"failure_category must be a FailureCategory, got {failure_category!r}" + ) + if next_retry_delay is not None and next_retry_delay < timedelta(0): + raise ValueError("next_retry_delay must not be negative") + message = f"{reason} details={list(details)!r}" if details else reason + super().__init__(message) + self.reason = reason + self.failure_category = failure_category + self.next_retry_delay = next_retry_delay + self._values: tuple[Any, ...] = details + self._payload: Payload | None = None + self._data_converter: "DataConverter | None" = None + + @staticmethod + def _from_payload( + reason: str, + payload: Payload, + data_converter: "DataConverter", + failure_category: FailureCategory | None = None, + next_retry_delay: timedelta | None = None, + ) -> "ApplicationError": + error = ApplicationError( + reason, + failure_category=failure_category, + next_retry_delay=next_retry_delay, ) - self.failure_details = failure_details - self.heartbeat_details = heartbeat_details + error._payload = payload + error._data_converter = data_converter + if payload.data: + error.message = ( + f"{reason}\n{payload.data.decode('utf-8', errors='replace')}" + ) + error.args = (error.message,) + return error + + def has_details(self) -> bool: + if self._payload is not None: + return bool(self._payload.data) + return bool(self._values) + + def details(self, *type_hints: type | None) -> list[Any]: + """Return the details of this failure. + + For a failure received from Cadence, the details are decoded with the + data converter using ``type_hints``. Decoding fails if the failure was + not created from an ``ApplicationError`` (for example, an arbitrary + Python exception, whose details are its traceback text). For an + ``ApplicationError`` constructed locally, the original values are + returned and ``type_hints`` are ignored. + """ + if self._payload is None or self._data_converter is None: + return list(self._values) + return self._data_converter.from_data(self._payload, list(type_hints)) + + +class _EncodedDetails: + """Details that are either local values or a payload received from Cadence.""" + + def _init_details(self, values: tuple[Any, ...]) -> None: + self._values = values + self._payload: Payload | None = None + self._data_converter: "DataConverter | None" = None + + def _set_payload(self, payload: Payload, data_converter: "DataConverter") -> None: + self._payload = payload + self._data_converter = data_converter + + def has_details(self) -> bool: + if self._payload is not None: + return bool(self._payload.data) + return bool(self._values) + + def details(self, *type_hints: type | None) -> list[Any]: + """Return the details, decoded using ``type_hints``. + + Returns an empty list if there are no details. For an error constructed + locally, the original values are returned and ``type_hints`` are ignored. + """ + if self._payload is None or self._data_converter is None: + return list(self._values) + if not self._payload.data: + return [] + return self._data_converter.from_data(self._payload, list(type_hints)) + + +class CadenceTimeoutError(_EncodedDetails, FailureError): + """An operation timed out. ``timeout_type`` identifies which timeout was exceeded. + + For an activity ``HEARTBEAT`` timeout, :meth:`details` returns the details + passed to the activity's last successful heartbeat, so the workflow can + resume from that progress. + """ + + def __init__(self, timeout_type: TimeoutType, *details: Any) -> None: + super().__init__(timeout_type.name) + self.timeout_type = timeout_type + self._init_details(details) + + @staticmethod + def _from_payload( + timeout_type: TimeoutType, + payload: Payload, + data_converter: "DataConverter", + ) -> "CadenceTimeoutError": + error = CadenceTimeoutError(timeout_type) + error._set_payload(payload, data_converter) + return error + +class WorkflowCanceledError(_EncodedDetails, FailureError): + """A workflow was canceled. :meth:`details` returns the details it was canceled with. + + For a child workflow, this is the cause of a :class:`ChildWorkflowError` when + something other than the parent canceled it. If the parent requested the + cancellation, :class:`asyncio.CancelledError` is raised instead. + """ -class ActivityCancelledError(Exception): def __init__(self, *details: Any) -> None: - super().__init__("Activity cancelled") - self.details = list(details) + super().__init__("canceled") + self._init_details(details) + @staticmethod + def _from_payload( + payload: Payload, data_converter: "DataConverter" + ) -> "WorkflowCanceledError": + error = WorkflowCanceledError() + error._set_payload(payload, data_converter) + return error -class WorkflowFailure(Exception): - def __init__(self, message: str) -> None: - super().__init__(message) +class TerminatedError(FailureError): + """A workflow was terminated.""" -class SignalFailure(Exception): - def __init__(self, message: str | None, signal_name: str) -> None: - if message is None: - message = f"Signal {signal_name} failed" - super().__init__(message) - self.signal_name = signal_name - self.message = message + def __init__(self) -> None: + super().__init__("terminated") + + +class ActivityError(FailureError): + """An activity failed or timed out. + ``cause`` is an :class:`ApplicationError` if the activity failed, or a + :class:`CadenceTimeoutError` if it timed out. Cancellation is raised as + :class:`asyncio.CancelledError` instead. + """ -class SignalExternalWorkflowFailed(Exception): - def __init__(self, message: str, cause: Any) -> None: + def __init__(self, message: str, *, activity_id: str, activity_type: str) -> None: super().__init__(message) - self.cause = cause + self.activity_id = activity_id + self.activity_type = activity_type + + @property + def cause(self) -> BaseException | None: + return self.__cause__ -class ChildWorkflowError(Exception): - """Base class for child workflow lifecycle errors.""" +class SignalExternalWorkflowError(FailureError): + """A signal sent from a workflow could not be delivered. + ``failure_cause`` is the ``SignalExternalWorkflowExecutionFailedCause`` + reported by Cadence, such as ``UNKNOWN_EXTERNAL_WORKFLOW_EXECUTION``. + """ -class StartChildWorkflowExecutionFailed(ChildWorkflowError): - def __init__(self, message: str, cause: Any, workflow_id: str) -> None: + def __init__( + self, + message: str, + failure_cause: int, + *, + workflow_id: str, + signal_name: str, + ) -> None: super().__init__(message) - self.cause = cause + self.failure_cause = failure_cause self.workflow_id = workflow_id + self.signal_name = signal_name -class ChildWorkflowExecutionFailed(ChildWorkflowError): - def __init__(self, message: str, failure: Any) -> None: - super().__init__(message) - self.failure = failure +class ChildWorkflowError(FailureError): + """A child workflow failed, timed out, was terminated, or was canceled externally. + ``cause`` is an :class:`ApplicationError` if the child workflow failed, a + :class:`CadenceTimeoutError` if it timed out, a :class:`TerminatedError` + if it was terminated, or a :class:`WorkflowCanceledError` if something other than + this workflow canceled it. If this workflow requested the cancellation, + :class:`asyncio.CancelledError` is raised instead. + """ -class ChildWorkflowExecutionCanceled(ChildWorkflowError): - def __init__(self, message: str, details: Any) -> None: + def __init__(self, message: str, *, workflow_id: str, workflow_type: str) -> None: super().__init__(message) - self.details = details + self.workflow_id = workflow_id + self.workflow_type = workflow_type + @property + def cause(self) -> BaseException | None: + return self.__cause__ -class ChildWorkflowExecutionTimedOut(ChildWorkflowError): - def __init__(self, message: str, timeout_type: int) -> None: - super().__init__(message) - self.timeout_type = timeout_type +class StartChildWorkflowExecutionError(ChildWorkflowError): + """A child workflow could not be started. -class ChildWorkflowExecutionTerminated(ChildWorkflowError): - pass + ``failure_cause`` is the ``ChildWorkflowExecutionFailedCause`` reported by + Cadence, such as ``WORKFLOW_ALREADY_RUNNING``. + """ + + def __init__( + self, + message: str, + failure_cause: int, + *, + workflow_id: str, + workflow_type: str, + ) -> None: + super().__init__(message, workflow_id=workflow_id, workflow_type=workflow_type) + self.failure_cause = failure_cause -class CadenceRpcError(Exception): +class CadenceRpcError(CadenceError): def __init__(self, message: str | None, code: grpc.StatusCode, *args): if message is None: super().__init__(code, *args) diff --git a/cadence/testing/_workflow_environment.py b/cadence/testing/_workflow_environment.py index 47b00f3..cad3d75 100644 --- a/cadence/testing/_workflow_environment.py +++ b/cadence/testing/_workflow_environment.py @@ -62,6 +62,10 @@ ) from cadence._internal.activity._definition import BaseDefinition +from cadence._internal.failure import ( + activity_error_from_exception, + child_workflow_error_from_exception, +) from cadence._internal.workflow.child_workflow_future import ChildWorkflowFutureImpl from cadence._internal.workflow.deterministic_event_loop import DeterministicEventLoop from cadence._internal.workflow.search_attributes import search_attributes_to_proto @@ -83,6 +87,7 @@ from cadence.client import Client, ClientOptions, StartWorkflowOptions, _identity from cadence.context import ContextPropagator from cadence.data_converter import DataConverter, DefaultDataConverter +from cadence.error import ContinueAsNewError from cadence.metrics import NoOpMetricsEmitter from cadence.workflow import ( ActivityOptions, @@ -486,10 +491,10 @@ def run_query(self, query_type: str, args: Payload) -> Payload: return self._data_converter.to_data([result]) def _collect_outcome(self) -> None: - signal_failure = self._instance.get_signal_failure() - if signal_failure is not None: + signal_error = self._instance.get_signal_handler_error() + if signal_error is not None: self.completed = True - self.error = signal_failure + self.error = signal_error[1] return if self._instance.is_done(): self.completed = True @@ -889,7 +894,11 @@ async def _body() -> Any: ) return result - result = await _run_in_fresh_context(_body) + try: + result = await _run_in_fresh_context(_body) + except Exception as e: + # Surface the failure the way a workflow sees it in production + raise activity_error_from_exception(e, name, str(uuid.uuid4()), dc) result_payload = dc.to_data([result]) return cast(ResultType, dc.from_data(result_payload, [result_type])[0]) @@ -935,9 +944,17 @@ async def _start_child_workflow( ) input_payload = self._data_converter.to_data(list(args)) - result_payload = await self._run_workflow_coro( - definition, child_info, input_payload, headers - ) + try: + result_payload = await self._run_workflow_coro( + definition, child_info, input_payload, headers + ) + except ContinueAsNewError: + raise + except Exception as e: + # Surface the failure the way a parent workflow sees it in production + raise child_workflow_error_from_exception( + e, definition.name, child_id, self._data_converter + ) loop = cast(DeterministicEventLoop, get_running_loop()) result_future: "Future[Payload]" = loop.create_future() diff --git a/tests/cadence/_internal/activity/test_activity_executor.py b/tests/cadence/_internal/activity/test_activity_executor.py index 74805d7..927717b 100644 --- a/tests/cadence/_internal/activity/test_activity_executor.py +++ b/tests/cadence/_internal/activity/test_activity_executor.py @@ -1,4 +1,5 @@ import asyncio +import logging import time from datetime import timedelta, datetime from unittest.mock import Mock, AsyncMock, PropertyMock @@ -11,6 +12,9 @@ from cadence._internal.activity import ActivityExecutor from cadence.activity import ActivityInfo, ActivityDefinition from cadence.api.v1.common_pb2 import ( + FAILURE_CATEGORY_FATAL, + FAILURE_CATEGORY_POLL, + FailureOptions, Header, WorkflowExecution, ActivityType, @@ -31,7 +35,7 @@ RecordActivityTaskHeartbeatRequest, ) from cadence.data_converter import DefaultDataConverter -from cadence.error import ActivityCancelledError +from cadence.error import ActivityCancelledError, ApplicationError, FailureCategory from cadence.worker import Registry @@ -100,6 +104,69 @@ async def activity_fn(): ) +async def test_activity_application_error_reports_reason_details_and_options( + client, +): + worker_stub = client.worker_stub + worker_stub.RespondActivityTaskFailed = AsyncMock( + return_value=RespondActivityTaskFailedResponse() + ) + + reg = Registry() + + @reg.activity(name="activity_type") + async def activity_fn(): + raise ApplicationError( + "test-error", + {"code": 7}, + "details", + failure_category=FailureCategory.FATAL, + next_retry_delay=timedelta(seconds=30), + ) + + executor = ActivityExecutor(client, "task_list", "identity", 1, reg.get_activity) + + await executor.execute(fake_task("activity_type", "")) + + worker_stub.RespondActivityTaskFailed.assert_called_once_with( + RespondActivityTaskFailedRequest( + task_token=b"task_token", + failure=Failure( + reason="test-error", + details=b'{"code":7} "details"', + options=FailureOptions( + failure_category=FAILURE_CATEGORY_FATAL, + next_retry_interval=from_timedelta(timedelta(seconds=30)), + ), + ), + identity="identity", + ) + ) + + +async def test_activity_poll_category_failure_not_logged_as_error(client, caplog): + worker_stub = client.worker_stub + worker_stub.RespondActivityTaskFailed = AsyncMock( + return_value=RespondActivityTaskFailedResponse() + ) + + reg = Registry() + + @reg.activity(name="activity_type") + async def activity_fn(): + raise ApplicationError("not-ready", failure_category=FailureCategory.POLL) + + executor = ActivityExecutor(client, "task_list", "identity", 1, reg.get_activity) + + with caplog.at_level(logging.DEBUG): + await executor.execute(fake_task("activity_type", "")) + + assert not [r for r in caplog.records if r.levelno >= logging.ERROR] + call = worker_stub.RespondActivityTaskFailed.call_args[0][0] + assert call.failure.reason == "not-ready" + assert call.failure.options.failure_category == FAILURE_CATEGORY_POLL + + async def test_activity_args(client): worker_stub = client.worker_stub worker_stub.RespondActivityTaskCompleted = AsyncMock( diff --git a/tests/cadence/_internal/test_failure.py b/tests/cadence/_internal/test_failure.py new file mode 100644 index 0000000..0e95818 --- /dev/null +++ b/tests/cadence/_internal/test_failure.py @@ -0,0 +1,129 @@ +from datetime import timedelta + +import pytest + +from cadence._internal.failure import ( + application_error_from_failure, + failure_from_exception, +) +from cadence.api.v1.common_pb2 import FAILURE_CATEGORY_POLL, Failure +from cadence.data_converter import DefaultDataConverter +from cadence.error import ApplicationError, FailureCategory + +DC = DefaultDataConverter() + + +def test_application_error_details_round_trip(): + error = ApplicationError("my-reason", {"a": 1}, [2, 3], "four") + + failure = failure_from_exception(error, DC, include_options=False) + received = application_error_from_failure(failure, DC) + + assert failure.reason == "my-reason" + assert received.reason == "my-reason" + assert received.has_details() + assert received.details(dict, list[int], str) == [{"a": 1}, [2, 3], "four"] + + +def test_application_error_without_details(): + failure = failure_from_exception( + ApplicationError("my-reason"), DC, include_options=False + ) + + assert failure == Failure(reason="my-reason") + assert not application_error_from_failure(failure, DC).has_details() + + +def test_other_exception_reason_is_class_name_and_details_are_text(): + def fail() -> None: + raise KeyError("missing") + + try: + fail() + except KeyError as e: + failure = failure_from_exception(e, DC, include_options=True) + + details = failure.details.decode() + assert failure.reason == "KeyError" + assert details.startswith("message: 'missing'\nstacktrace: ") + assert 'raise KeyError("missing")' in details + assert not failure.HasField("options") + + +def test_received_application_error_is_passed_through_unchanged(): + original = Failure(reason="from-go", details=b'{"Go":"encoded"}') + received = application_error_from_failure(original, DC) + + assert failure_from_exception(received, DC, include_options=False) == original + + +def test_failure_options_only_included_when_requested(): + error = ApplicationError( + "my-reason", + failure_category=FailureCategory.POLL, + next_retry_delay=timedelta(seconds=5), + ) + + with_options = failure_from_exception(error, DC, include_options=True) + without_options = failure_from_exception(error, DC, include_options=False) + + assert with_options.options.failure_category == FAILURE_CATEGORY_POLL + assert with_options.options.next_retry_interval.ToTimedelta() == timedelta( + seconds=5 + ) + assert not without_options.HasField("options") + + +def test_no_failure_options_when_unset(): + failure = failure_from_exception( + ApplicationError("my-reason"), DC, include_options=True + ) + + assert not failure.HasField("options") + + +def test_next_retry_delay_only(): + failure = failure_from_exception( + ApplicationError("my-reason", next_retry_delay=timedelta(seconds=3)), + DC, + include_options=True, + ) + + assert failure.options.next_retry_interval.ToTimedelta() == timedelta(seconds=3) + received = application_error_from_failure(failure, DC) + assert received.failure_category is None + assert received.next_retry_delay == timedelta(seconds=3) + + +def test_next_retry_delay_rounds_up_to_whole_seconds(): + failure = failure_from_exception( + ApplicationError("my-reason", next_retry_delay=timedelta(seconds=1.2)), + DC, + include_options=True, + ) + + assert failure.options.next_retry_interval.ToTimedelta() == timedelta(seconds=2) + + +def test_unencodable_details_keep_reason(): + failure = failure_from_exception( + ApplicationError("my-reason", object()), DC, include_options=False + ) + + assert failure.reason == "my-reason" + assert b"failed to encode ApplicationError details" in failure.details + + +def test_received_application_error_str_includes_details_text(): + received = application_error_from_failure( + Failure(reason="RuntimeError", details=b"message: boom"), DC + ) + + assert str(received) == "RuntimeError\nmessage: boom" + + +def test_application_error_validation(): + with pytest.raises(ValueError): + ApplicationError("my-reason", next_retry_delay=timedelta(seconds=-1)) + with pytest.raises(TypeError): + ApplicationError("my-reason", failure_category=1) # type: ignore[arg-type] diff --git a/tests/cadence/_internal/workflow/statemachine/test_activity_state_machine.py b/tests/cadence/_internal/workflow/statemachine/test_activity_state_machine.py index 9590558..bd82244 100644 --- a/tests/cadence/_internal/workflow/statemachine/test_activity_state_machine.py +++ b/tests/cadence/_internal/workflow/statemachine/test_activity_state_machine.py @@ -1,6 +1,8 @@ from asyncio import CancelledError +from datetime import timedelta import pytest +from google.protobuf.duration_pb2 import Duration from cadence._internal.workflow.statemachine.activity_state_machine import ( ActivityStateMachine, @@ -12,10 +14,23 @@ record_immediate_cancel, ) from cadence.api.v1 import decision, history -from cadence.api.v1.common_pb2 import Payload, Failure +from cadence.api.v1.common_pb2 import ( + FAILURE_CATEGORY_POLL, + ActivityType, + Failure, + FailureOptions, + Payload, +) from cadence.api.v1.decision_pb2 import RequestCancelActivityTaskDecisionAttributes from cadence.api.v1.workflow_pb2 import TimeoutType -from cadence.error import ActivityFailure +from cadence.data_converter import DefaultDataConverter +from cadence.error import ( + ActivityError, + ApplicationError, + CadenceTimeoutError, + FailureCategory, +) +from cadence.error import TimeoutType as CadenceTimeoutType ### These tests have to be async because they rely on the presence of an eventloop @@ -23,7 +38,7 @@ async def test_activity_state_machine_initiated(): attrs = decision.ScheduleActivityTaskDecisionAttributes(activity_id="a") completed = DecisionFuture[Payload]() - m = ActivityStateMachine(attrs, completed) + m = ActivityStateMachine(attrs, completed, DefaultDataConverter()) assert completed.done() is False assert m.get_decision() == decision.Decision( @@ -35,7 +50,7 @@ async def test_activity_state_machine_cancelled_before_initiated(): attrs = decision.ScheduleActivityTaskDecisionAttributes(activity_id="a") completed = DecisionFuture[Payload]() - m = ActivityStateMachine(attrs, completed) + m = ActivityStateMachine(attrs, completed, DefaultDataConverter()) res = m.request_cancel() assert res is True @@ -47,7 +62,7 @@ async def test_activity_state_machine_cancelled_before_initiated(): async def test_activity_state_machine_cancelled_after_initiated(): attrs = decision.ScheduleActivityTaskDecisionAttributes(activity_id="a") completed = DecisionFuture[Payload]() - m = ActivityStateMachine(attrs, completed) + m = ActivityStateMachine(attrs, completed, DefaultDataConverter()) m.handle_scheduled(history.ActivityTaskScheduledEventAttributes(activity_id="a")) res = m.request_cancel() @@ -64,7 +79,7 @@ async def test_activity_state_machine_cancelled_after_initiated(): async def test_activity_state_machine_completed(): attrs = decision.ScheduleActivityTaskDecisionAttributes(activity_id="a") completed = DecisionFuture[Payload]() - m = ActivityStateMachine(attrs, completed) + m = ActivityStateMachine(attrs, completed, DefaultDataConverter()) m.handle_scheduled(history.ActivityTaskScheduledEventAttributes(activity_id="a")) m.handle_started(history.ActivityTaskStartedEventAttributes()) @@ -78,16 +93,17 @@ async def test_activity_state_machine_completed(): async def test_activity_state_machine_timeout(): - attrs = decision.ScheduleActivityTaskDecisionAttributes(activity_id="a") + attrs = decision.ScheduleActivityTaskDecisionAttributes( + activity_id="a", activity_type=ActivityType(name="my_activity") + ) completed = DecisionFuture[Payload]() - m = ActivityStateMachine(attrs, completed) + m = ActivityStateMachine(attrs, completed, DefaultDataConverter()) m.handle_scheduled(history.ActivityTaskScheduledEventAttributes(activity_id="a")) - heartbeat_details = Payload(data=b'{"progress": 42}') m.handle_timeout( history.ActivityTaskTimedOutEventAttributes( timeout_type=TimeoutType.TIMEOUT_TYPE_HEARTBEAT, - details=heartbeat_details, + details=Payload(data=b'{"progress": 42}'), last_failure=Failure( reason="RuntimeError", details=b"Traceback (most recent call last):\nRuntimeError: boom", @@ -97,19 +113,45 @@ async def test_activity_state_machine_timeout(): assert completed.done() is True assert m.get_decision() is None - with pytest.raises(ActivityFailure, match="TIMEOUT_TYPE_HEARTBEAT") as exc_info: + with pytest.raises(ActivityError, match="HEARTBEAT") as exc_info: completed.result() - assert exc_info.value.failure_details == ( - "Traceback (most recent call last):\nRuntimeError: boom" + error = exc_info.value + assert error.activity_id == "a" + assert error.activity_type == "my_activity" + timeout = error.cause + assert isinstance(timeout, CadenceTimeoutError) + assert timeout.timeout_type is CadenceTimeoutType.HEARTBEAT + assert timeout.has_details() + assert timeout.details(dict) == [{"progress": 42}] + + +async def test_activity_state_machine_timeout_without_details(): + attrs = decision.ScheduleActivityTaskDecisionAttributes( + activity_id="a", activity_type=ActivityType(name="my_activity") ) - assert exc_info.value.heartbeat_details == heartbeat_details - assert exc_info.value.heartbeat_details.data == b'{"progress": 42}' + completed = DecisionFuture[Payload]() + m = ActivityStateMachine(attrs, completed, DefaultDataConverter()) + + m.handle_scheduled(history.ActivityTaskScheduledEventAttributes(activity_id="a")) + m.handle_timeout( + history.ActivityTaskTimedOutEventAttributes( + timeout_type=TimeoutType.TIMEOUT_TYPE_START_TO_CLOSE, + ) + ) + + with pytest.raises(ActivityError) as exc_info: + completed.result() + timeout = exc_info.value.cause + assert isinstance(timeout, CadenceTimeoutError) + assert timeout.timeout_type is CadenceTimeoutType.START_TO_CLOSE + assert not timeout.has_details() + assert timeout.details(dict) == [] async def test_activity_state_machine_failed(): attrs = decision.ScheduleActivityTaskDecisionAttributes(activity_id="a") completed = DecisionFuture[Payload]() - m = ActivityStateMachine(attrs, completed) + m = ActivityStateMachine(attrs, completed, DefaultDataConverter()) m.handle_scheduled(history.ActivityTaskScheduledEventAttributes(activity_id="a")) m.handle_started(history.ActivityTaskStartedEventAttributes()) @@ -124,19 +166,46 @@ async def test_activity_state_machine_failed(): assert completed.done() is True assert m.get_decision() is None - with pytest.raises( - ActivityFailure, match="RuntimeError.*error message" - ) as exc_info: + with pytest.raises(ActivityError, match="RuntimeError.*error message") as exc_info: completed.result() - assert exc_info.value.failure_details == ( - "Traceback (most recent call last):\nRuntimeError: error message" + cause = exc_info.value.cause + assert isinstance(cause, ApplicationError) + assert cause.reason == "RuntimeError" + + +async def test_activity_state_machine_failed_with_application_error(): + attrs = decision.ScheduleActivityTaskDecisionAttributes(activity_id="a") + completed = DecisionFuture[Payload]() + m = ActivityStateMachine(attrs, completed, DefaultDataConverter()) + + m.handle_scheduled(history.ActivityTaskScheduledEventAttributes(activity_id="a")) + m.handle_failed( + history.ActivityTaskFailedEventAttributes( + failure=Failure( + reason="not-ready", + details=b'{"retry_in": 5} "later"', + options=FailureOptions( + failure_category=FAILURE_CATEGORY_POLL, + next_retry_interval=Duration(seconds=5), + ), + ) + ) ) + with pytest.raises(ActivityError) as exc_info: + completed.result() + cause = exc_info.value.cause + assert isinstance(cause, ApplicationError) + assert cause.reason == "not-ready" + assert cause.details(dict, str) == [{"retry_in": 5}, "later"] + assert cause.failure_category is FailureCategory.POLL + assert cause.next_retry_delay == timedelta(seconds=5) + async def test_activity_state_machine_cancel_confirmed(): attrs = decision.ScheduleActivityTaskDecisionAttributes(activity_id="a") completed = DecisionFuture[Payload]() - m = ActivityStateMachine(attrs, completed) + m = ActivityStateMachine(attrs, completed, DefaultDataConverter()) m.handle_scheduled(history.ActivityTaskScheduledEventAttributes(activity_id="a")) m.request_cancel() @@ -150,7 +219,7 @@ async def test_activity_state_machine_cancel_confirmed(): async def test_activity_state_machine_complete_after_cancel(): attrs = decision.ScheduleActivityTaskDecisionAttributes(activity_id="a") completed = DecisionFuture[Payload]() - m = ActivityStateMachine(attrs, completed) + m = ActivityStateMachine(attrs, completed, DefaultDataConverter()) m.handle_scheduled(history.ActivityTaskScheduledEventAttributes(activity_id="a")) m.request_cancel() @@ -167,7 +236,7 @@ async def test_activity_state_machine_complete_after_cancel(): async def test_activity_state_machine_cancel_accepted(): attrs = decision.ScheduleActivityTaskDecisionAttributes(activity_id="a") completed = DecisionFuture[Payload]() - m = ActivityStateMachine(attrs, completed) + m = ActivityStateMachine(attrs, completed, DefaultDataConverter()) m.handle_scheduled(history.ActivityTaskScheduledEventAttributes(activity_id="a")) m.request_cancel() @@ -188,7 +257,7 @@ async def test_activity_state_machine_cancel_accepted(): async def test_activity_state_machine_cancel_failed(): attrs = decision.ScheduleActivityTaskDecisionAttributes(activity_id="a") completed = DecisionFuture[Payload]() - m = ActivityStateMachine(attrs, completed) + m = ActivityStateMachine(attrs, completed, DefaultDataConverter()) m.handle_scheduled(history.ActivityTaskScheduledEventAttributes(activity_id="a")) m.request_cancel() @@ -203,7 +272,7 @@ async def test_activity_state_machine_cancel_failed(): async def test_activity_state_machine_completed_after_cancel_failed(): attrs = decision.ScheduleActivityTaskDecisionAttributes(activity_id="a") completed = DecisionFuture[Payload]() - m = ActivityStateMachine(attrs, completed) + m = ActivityStateMachine(attrs, completed, DefaultDataConverter()) m.handle_scheduled(history.ActivityTaskScheduledEventAttributes(activity_id="a")) m.request_cancel() diff --git a/tests/cadence/_internal/workflow/statemachine/test_child_workflow_execution_state_machine.py b/tests/cadence/_internal/workflow/statemachine/test_child_workflow_execution_state_machine.py index d02b6f8..c38d743 100644 --- a/tests/cadence/_internal/workflow/statemachine/test_child_workflow_execution_state_machine.py +++ b/tests/cadence/_internal/workflow/statemachine/test_child_workflow_execution_state_machine.py @@ -1,14 +1,18 @@ +from asyncio import CancelledError + import pytest from cadence._internal.workflow.statemachine.child_workflow_execution_state_machine import ( ChildWorkflowExecutionStateMachine, ) from cadence.error import ( - ChildWorkflowExecutionCanceled, - ChildWorkflowExecutionFailed, - ChildWorkflowExecutionTerminated, - ChildWorkflowExecutionTimedOut, - StartChildWorkflowExecutionFailed, + ApplicationError, + CadenceTimeoutError, + WorkflowCanceledError, + ChildWorkflowError, + StartChildWorkflowExecutionError, + TerminatedError, + TimeoutType, ) from cadence._internal.workflow.statemachine.decision_state_machine import ( DecisionFuture, @@ -19,6 +23,11 @@ ) from cadence.api.v1 import decision, history from cadence.api.v1.common_pb2 import Failure, Payload, WorkflowExecution, WorkflowType +from cadence.api.v1.workflow_pb2 import ( + CHILD_WORKFLOW_EXECUTION_FAILED_CAUSE_WORKFLOW_ALREADY_RUNNING, + TIMEOUT_TYPE_START_TO_CLOSE, +) +from cadence.data_converter import DefaultDataConverter ### These tests have to be async because they rely on the presence of an eventloop @@ -38,7 +47,9 @@ def make_sm() -> tuple[ ) execution: DecisionFuture[WorkflowExecution] = DecisionFuture() result: DecisionFuture[Payload] = DecisionFuture() - sm = ChildWorkflowExecutionStateMachine(attrs, execution, result) + sm = ChildWorkflowExecutionStateMachine( + attrs, execution, result, DefaultDataConverter() + ) return sm, execution, result @@ -172,16 +183,25 @@ async def test_handle_initiation_failed_reuses_same_exception(): sm.handle_initiation_failed( history.StartChildWorkflowExecutionFailedEventAttributes( workflow_id=WF_ID, + cause=CHILD_WORKFLOW_EXECUTION_FAILED_CAUSE_WORKFLOW_ALREADY_RUNNING, ) ) assert sm.state is DecisionState.COMPLETED assert execution.done() is True assert result.done() is True - with pytest.raises(StartChildWorkflowExecutionFailed, match="start child failed"): + with pytest.raises(StartChildWorkflowExecutionError): execution.result() - with pytest.raises(StartChildWorkflowExecutionFailed, match="start child failed"): + with pytest.raises( + StartChildWorkflowExecutionError, match="WORKFLOW_ALREADY_RUNNING$" + ) as exc_info: result.result() + assert exc_info.value.workflow_id == WF_ID + assert exc_info.value.workflow_type == "MyWorkflow" + assert ( + exc_info.value.failure_cause + == CHILD_WORKFLOW_EXECUTION_FAILED_CAUSE_WORKFLOW_ALREADY_RUNNING + ) # Both futures hold the identical exception object (not just equal ones). exc_execution = execution.exception() @@ -235,16 +255,48 @@ async def test_handle_failed(): sm.handle_failed( history.ChildWorkflowExecutionFailedEventAttributes( - failure=Failure(reason="boom") + workflow_type=WorkflowType(name="MyWorkflow"), + workflow_execution=WorkflowExecution(workflow_id=WF_ID, run_id="run-1"), + failure=Failure(reason="boom", details=b'{"code": 7}'), + ) + ) + + assert sm.state is DecisionState.COMPLETED + with pytest.raises(ChildWorkflowError, match="boom") as exc_info: + result.result() + assert exc_info.value.workflow_id == WF_ID + assert exc_info.value.workflow_type == "MyWorkflow" + cause = exc_info.value.cause + assert isinstance(cause, ApplicationError) + assert cause.reason == "boom" + assert cause.details(dict) == [{"code": 7}] + + +async def test_handle_canceled_after_cancel_requested_raises_cancelled_error(): + sm, execution, result = make_sm() + sm.handle_initiated(history.StartChildWorkflowExecutionInitiatedEventAttributes()) + sm.handle_started( + history.ChildWorkflowExecutionStartedEventAttributes( + workflow_execution=WorkflowExecution(workflow_id=WF_ID, run_id="run-1") + ) + ) + sm.request_cancel() + sm.handle_cancel_initiated( + history.RequestCancelExternalWorkflowExecutionInitiatedEventAttributes() + ) + + sm.handle_canceled( + history.ChildWorkflowExecutionCanceledEventAttributes( + details=Payload(data=b'"cancel-details"') ) ) assert sm.state is DecisionState.COMPLETED - with pytest.raises(ChildWorkflowExecutionFailed, match="boom"): + with pytest.raises(CancelledError, match="cancel-details"): result.result() -async def test_handle_canceled(): +async def test_handle_canceled_externally_raises_child_workflow_error(): sm, execution, result = make_sm() sm.handle_initiated(history.StartChildWorkflowExecutionInitiatedEventAttributes()) sm.handle_started( @@ -255,13 +307,44 @@ async def test_handle_canceled(): sm.handle_canceled( history.ChildWorkflowExecutionCanceledEventAttributes( - details=Payload(data=b"cancel-details") + workflow_type=WorkflowType(name="MyWorkflow"), + workflow_execution=WorkflowExecution(workflow_id=WF_ID, run_id="run-1"), + details=Payload(data=b'"cancel-details"'), ) ) assert sm.state is DecisionState.COMPLETED - with pytest.raises(ChildWorkflowExecutionCanceled, match="child workflow canceled"): + with pytest.raises(ChildWorkflowError, match="child workflow canceled") as exc_info: result.result() + assert exc_info.value.workflow_id == WF_ID + assert exc_info.value.workflow_type == "MyWorkflow" + cause = exc_info.value.cause + assert isinstance(cause, WorkflowCanceledError) + assert cause.details(str) == ["cancel-details"] + + +async def test_handle_canceled_after_cancel_failed_raises_child_workflow_error(): + sm, execution, result = make_sm() + sm.handle_initiated(history.StartChildWorkflowExecutionInitiatedEventAttributes()) + sm.handle_started( + history.ChildWorkflowExecutionStartedEventAttributes( + workflow_execution=WorkflowExecution(workflow_id=WF_ID, run_id="run-1") + ) + ) + sm.request_cancel() + sm.handle_cancel_initiated( + history.RequestCancelExternalWorkflowExecutionInitiatedEventAttributes() + ) + sm.handle_cancel_failed( + history.RequestCancelExternalWorkflowExecutionFailedEventAttributes() + ) + + sm.handle_canceled(history.ChildWorkflowExecutionCanceledEventAttributes()) + + with pytest.raises(ChildWorkflowError) as exc_info: + result.result() + assert isinstance(exc_info.value.cause, WorkflowCanceledError) + assert not exc_info.value.cause.has_details() async def test_handle_timed_out(): @@ -273,13 +356,22 @@ async def test_handle_timed_out(): ) ) - sm.handle_timed_out(history.ChildWorkflowExecutionTimedOutEventAttributes()) + sm.handle_timed_out( + history.ChildWorkflowExecutionTimedOutEventAttributes( + workflow_type=WorkflowType(name="MyWorkflow"), + workflow_execution=WorkflowExecution(workflow_id=WF_ID, run_id="run-1"), + timeout_type=TIMEOUT_TYPE_START_TO_CLOSE, + ) + ) assert sm.state is DecisionState.COMPLETED with pytest.raises( - ChildWorkflowExecutionTimedOut, match="child workflow timed out" - ): + ChildWorkflowError, match="child workflow timed out" + ) as exc_info: result.result() + cause = exc_info.value.cause + assert isinstance(cause, CadenceTimeoutError) + assert cause.timeout_type is TimeoutType.START_TO_CLOSE async def test_handle_terminated(): @@ -291,11 +383,19 @@ async def test_handle_terminated(): ) ) - sm.handle_terminated(history.ChildWorkflowExecutionTerminatedEventAttributes()) + sm.handle_terminated( + history.ChildWorkflowExecutionTerminatedEventAttributes( + workflow_type=WorkflowType(name="MyWorkflow"), + workflow_execution=WorkflowExecution(workflow_id=WF_ID, run_id="run-1"), + ) + ) assert sm.state is DecisionState.COMPLETED - with pytest.raises(ChildWorkflowExecutionTerminated): + with pytest.raises( + ChildWorkflowError, match="child workflow terminated" + ) as exc_info: result.result() + assert isinstance(exc_info.value.cause, TerminatedError) async def test_cancel_initiated_transitions_to_cancellation_recorded(): diff --git a/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py b/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py index 6c16d44..ca6e8f2 100644 --- a/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py +++ b/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py @@ -3,7 +3,11 @@ import pytest -from cadence.error import ChildWorkflowError, StartChildWorkflowExecutionFailed +from cadence.error import ( + ChildWorkflowError, + SignalExternalWorkflowError, + StartChildWorkflowExecutionError, +) from cadence._internal.workflow.statemachine.cancellation import CANCEL_MARKER_NAME from cadence._internal.workflow.statemachine.decision_manager import DecisionManager from cadence._internal.workflow.statemachine.event_dispatcher import ( @@ -16,9 +20,6 @@ mutable_side_effect_marker_info, ) from cadence._internal.workflow.statemachine.nondeterminism import NonDeterminismError -from cadence._internal.workflow.statemachine.signal_external_workflow_state_machine import ( - SignalExternalWorkflowFailed, -) from cadence.api.v1 import history, decision from cadence.api.v1.common_pb2 import Payload, WorkflowExecution, WorkflowType @@ -556,7 +557,7 @@ async def test_child_workflow_initiation_failed_dispatch(): assert execution.done() is True assert result.done() is True - with pytest.raises(StartChildWorkflowExecutionFailed): + with pytest.raises(StartChildWorkflowExecutionError): result.result() @@ -843,7 +844,7 @@ async def test_signal_external_workflow_failed_dispatch(): decisions.handle_history_event(signal_failed(2, initiated_event_id=1)) assert future.done() is True - with pytest.raises(SignalExternalWorkflowFailed): + with pytest.raises(SignalExternalWorkflowError): future.result() diff --git a/tests/cadence/_internal/workflow/statemachine/test_signal_external_workflow_state_machine.py b/tests/cadence/_internal/workflow/statemachine/test_signal_external_workflow_state_machine.py index 0f4e092..edd8d0d 100644 --- a/tests/cadence/_internal/workflow/statemachine/test_signal_external_workflow_state_machine.py +++ b/tests/cadence/_internal/workflow/statemachine/test_signal_external_workflow_state_machine.py @@ -7,9 +7,12 @@ from cadence._internal.workflow.statemachine.signal_external_workflow_state_machine import ( SignalExternalWorkflowStateMachine, ) -from cadence.error import SignalExternalWorkflowFailed +from cadence.error import SignalExternalWorkflowError from cadence.api.v1 import decision, history from cadence.api.v1.common_pb2 import WorkflowExecution +from cadence.api.v1.workflow_pb2 import ( + SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_UNKNOWN_EXTERNAL_WORKFLOW_EXECUTION, +) SIGNAL_ID = "0" DOMAIN = "test-domain" @@ -90,12 +93,21 @@ async def test_failed_rejects_future(): sm.handle_failed( history.SignalExternalWorkflowExecutionFailedEventAttributes( initiated_event_id=1, + cause=SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_UNKNOWN_EXTERNAL_WORKFLOW_EXECUTION, ) ) assert sm.state is DecisionState.COMPLETED assert completed.done() is True with pytest.raises( - SignalExternalWorkflowFailed, match="signal external workflow failed" - ): + SignalExternalWorkflowError, + match=r"signal_name=my_signal, workflow_id=child-wf-1\): " + r"UNKNOWN_EXTERNAL_WORKFLOW_EXECUTION$", + ) as exc_info: completed.result() + assert exc_info.value.workflow_id == "child-wf-1" + assert exc_info.value.signal_name == "my_signal" + assert ( + exc_info.value.failure_cause + == SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_UNKNOWN_EXTERNAL_WORKFLOW_EXECUTION + ) diff --git a/tests/cadence/_internal/workflow/test_context_child_workflow.py b/tests/cadence/_internal/workflow/test_context_child_workflow.py index f0f98ed..7dda447 100644 --- a/tests/cadence/_internal/workflow/test_context_child_workflow.py +++ b/tests/cadence/_internal/workflow/test_context_child_workflow.py @@ -12,7 +12,7 @@ from cadence.api.v1.decision_pb2 import StartChildWorkflowExecutionDecisionAttributes from cadence.api.v1.tasklist_pb2 import TaskList, TaskListKind from cadence.data_converter import DefaultDataConverter -from cadence.error import ChildWorkflowExecutionFailed +from cadence.error import ChildWorkflowError from cadence.workflow import WorkflowInfo @@ -304,10 +304,12 @@ async def test_execute_child_workflow_deserializes_result(): @pytest.mark.asyncio async def test_execute_child_workflow_propagates_errors(): ctx, dm = _make_ctx() - failure = ChildWorkflowExecutionFailed("child failed", failure=None) + failure = ChildWorkflowError( + "child failed", workflow_id="child-1", workflow_type="ChildWf" + ) _setup_schedule_mock_error(dm, failure) - with pytest.raises(ChildWorkflowExecutionFailed, match="child failed"): + with pytest.raises(ChildWorkflowError, match="child failed"): await ctx.execute_child_workflow( "ChildWf", str, @@ -439,7 +441,9 @@ async def test_start_child_workflow_future_propagates_errors(): ) result_future: asyncio.Future = loop.create_future() result_future.set_exception( - ChildWorkflowExecutionFailed("child failed", failure=None) + ChildWorkflowError( + "child failed", workflow_id="child-1", workflow_type="ChildWf" + ) ) dm.schedule_child_workflow = MagicMock( return_value=(execution_future, result_future) @@ -450,5 +454,5 @@ async def test_start_child_workflow_future_propagates_errors(): str, execution_start_to_close_timeout=timedelta(minutes=5), ) - with pytest.raises(ChildWorkflowExecutionFailed, match="child failed"): + with pytest.raises(ChildWorkflowError, match="child failed"): await future diff --git a/tests/cadence/_internal/workflow/test_signal_handling.py b/tests/cadence/_internal/workflow/test_signal_handling.py index 5baa145..3f9aece 100644 --- a/tests/cadence/_internal/workflow/test_signal_handling.py +++ b/tests/cadence/_internal/workflow/test_signal_handling.py @@ -18,9 +18,9 @@ WorkflowExecutionStartedEventAttributes, WorkflowExecutionSignaledEventAttributes, ) +from cadence._internal.workflow.deterministic_event_loop import FatalDecisionError from cadence._internal.workflow.workflow_engine import WorkflowEngine from cadence.data_converter import DefaultDataConverter -from cadence.error import SignalFailure from cadence.workflow import WorkflowInfo, WorkflowDefinition, WorkflowDefinitionOptions @@ -643,11 +643,11 @@ def test_signal_handler_exception_fails_decision(self): ), ] - with pytest.raises(SignalFailure) as err: + with pytest.raises(FatalDecisionError, match="bad_signal") as err: engine.process_decision(events) - assert err.value.signal_name == "bad_signal" - assert "handler failed on: boom" in str(err.value) + assert isinstance(err.value.__cause__, ValueError) + assert "handler failed on: boom" in str(err.value.__cause__) def test_signal_with_invalid_payload_drops_signal(self, caplog): """Malformed signal payloads are logged as warnings and dropped.""" diff --git a/tests/cadence/_internal/workflow/test_workflow_engine.py b/tests/cadence/_internal/workflow/test_workflow_engine.py index 46d2862..3e75883 100644 --- a/tests/cadence/_internal/workflow/test_workflow_engine.py +++ b/tests/cadence/_internal/workflow/test_workflow_engine.py @@ -4,9 +4,10 @@ from typing import Any, List, Mapping import pytest -from cadence.api.v1.common_pb2 import ActivityType, Payload, WorkflowType +from cadence.api.v1.common_pb2 import ActivityType, Failure, Payload, WorkflowType from cadence.api.v1.decision_pb2 import CancelWorkflowExecutionDecisionAttributes from cadence.api.v1.history_pb2 import ( + ActivityTaskFailedEventAttributes, ActivityTaskScheduledEventAttributes, DecisionTaskCompletedEventAttributes, DecisionTaskScheduledEventAttributes, @@ -22,6 +23,7 @@ from cadence._internal.workflow.workflow_engine import WorkflowEngine from cadence import workflow from cadence.data_converter import DefaultDataConverter +from cadence.error import ActivityError, ApplicationError from cadence.workflow import ( SearchAttributeType, WorkflowContext, @@ -582,6 +584,137 @@ def test_upsert_search_attributes_replay_does_not_reemit(self): assert decision_result.decisions == [] + def test_workflow_application_error_fails_with_reason_and_encoded_details(self): + workflow_engine = create_workflow_engine( + WorkflowDefinition.wrap( + ApplicationErrorWorkflow, + WorkflowDefinitionOptions(name="application_error_workflow"), + ) + ) + + decision_result = workflow_engine.process_decision(_started_history()) + + assert len(decision_result.decisions) == 1 + failure = decision_result.decisions[ + 0 + ].fail_workflow_execution_decision_attributes.failure + assert failure == Failure(reason="my-reason", details=b'{"code":7} "text"') + assert not failure.HasField("options") + + def test_reraised_activity_failure_fails_workflow_with_same_failure(self): + workflow_engine = create_workflow_engine( + WorkflowDefinition.wrap( + ReraiseActivityCauseWorkflow, + WorkflowDefinitionOptions(name="reraise_activity_cause_workflow"), + ) + ) + activity_failure = Failure(reason="from-activity", details=b'{"code":7}') + + decision_result = workflow_engine.process_decision( + _history_with_activity_failure(activity_failure) + ) + + assert len(decision_result.decisions) == 1 + assert ( + decision_result.decisions[ + 0 + ].fail_workflow_execution_decision_attributes.failure + == activity_failure + ) + + def test_uncaught_activity_error_fails_workflow_with_wrapper_reason(self): + workflow_engine = create_workflow_engine( + WorkflowDefinition.wrap( + PendingActivityCancellationWorkflow, + WorkflowDefinitionOptions(name="uncaught_activity_error_workflow"), + ) + ) + + decision_result = workflow_engine.process_decision( + _history_with_activity_failure(Failure(reason="from-activity")) + ) + + assert len(decision_result.decisions) == 1 + failure = decision_result.decisions[ + 0 + ].fail_workflow_execution_decision_attributes.failure + assert failure.reason == "ActivityError" + assert "from-activity" in failure.details.decode() + + +class ApplicationErrorWorkflow: + @workflow.run + async def run(self): + raise ApplicationError("my-reason", {"code": 7}, "text") + + +class ReraiseActivityCauseWorkflow: + @workflow.run + async def run(self): + try: + await workflow.execute_activity( + "act", str, schedule_to_close_timeout=timedelta(minutes=5) + ) + except ActivityError as e: + assert e.cause is not None + raise e.cause + return "unreachable" + + +def _started_history() -> list[HistoryEvent]: + return [ + _event( + 1, + workflow_execution_started_event_attributes=WorkflowExecutionStartedEventAttributes(), + ), + _event( + 2, + decision_task_scheduled_event_attributes=DecisionTaskScheduledEventAttributes(), + ), + _event( + 3, + decision_task_started_event_attributes=DecisionTaskStartedEventAttributes( + scheduled_event_id=2 + ), + ), + ] + + +def _history_with_activity_failure(failure: Failure) -> list[HistoryEvent]: + return [ + *_started_history(), + _event( + 4, + decision_task_completed_event_attributes=DecisionTaskCompletedEventAttributes( + scheduled_event_id=2, started_event_id=3 + ), + ), + _event( + 5, + activity_task_scheduled_event_attributes=ActivityTaskScheduledEventAttributes( + activity_id="0", + activity_type=ActivityType(name="act"), + ), + ), + _event( + 6, + activity_task_failed_event_attributes=ActivityTaskFailedEventAttributes( + scheduled_event_id=5, + failure=failure, + ), + ), + _event( + 7, + decision_task_scheduled_event_attributes=DecisionTaskScheduledEventAttributes(), + ), + _event( + 8, + decision_task_started_event_attributes=DecisionTaskStartedEventAttributes( + scheduled_event_id=7 + ), + ), + ] + def create_workflow_engine( workflow_definition: WorkflowDefinition, diff --git a/tests/cadence/test_context.py b/tests/cadence/test_context.py index 1a951a3..1d2b89f 100644 --- a/tests/cadence/test_context.py +++ b/tests/cadence/test_context.py @@ -37,6 +37,7 @@ ) from cadence.context import ContextVarPropagator from cadence.data_converter import DefaultDataConverter +from cadence.error import ActivityError, ApplicationError from cadence.metrics import NoOpMetricsEmitter from cadence.testing import TestWorkflowEnvironment from cadence.worker._types import WorkerOptions @@ -487,8 +488,10 @@ async def run(self) -> str: with TestWorkflowEnvironment(registry) as environment: await environment.client.start_workflow("Parent", task_list="test") - with pytest.raises(LookupError): + with pytest.raises(ActivityError) as exc_info: environment.get_workflow_result(str) + assert isinstance(exc_info.value.cause, ApplicationError) + assert exc_info.value.cause.reason == "LookupError" def _started_events(headers: Mapping[str, bytes]) -> list[HistoryEvent]: diff --git a/tests/cadence/testing/test_workflow_environment.py b/tests/cadence/testing/test_workflow_environment.py index 9b29560..2a39b5c 100644 --- a/tests/cadence/testing/test_workflow_environment.py +++ b/tests/cadence/testing/test_workflow_environment.py @@ -6,6 +6,7 @@ import pytest from cadence import workflow +from cadence.error import ActivityError, ApplicationError, ChildWorkflowError from cadence.testing import TestWorkflowEnvironment from cadence.worker import Registry from cadence.workflow import ( @@ -85,6 +86,10 @@ def approve(self, value: str) -> None: def status(self) -> str: return "approved" if self._approved else "pending" + @workflow.signal(name="reject") + def reject(self, reason: str) -> None: + raise ValueError(f"rejected: {reason}") + @registry.workflow class TimerWorkflow: @@ -141,6 +146,34 @@ async def run(self) -> str: raise ValueError("boom") +@registry.workflow +class CatchActivityErrorWorkflow: + @workflow.run + async def run(self) -> str: + try: + await workflow.execute_activity( + "greet", + str, + "dave", + schedule_to_close_timeout=timedelta(seconds=10), + ) + except ActivityError as e: + assert isinstance(e.cause, ApplicationError) + return f"{e.activity_type}:{e.cause.reason}:{e.cause.details(dict)[0]}" + return "unreachable" + + +@registry.workflow +class FailingChildParentWorkflow: + @workflow.run + async def run(self) -> str: + return await workflow.execute_child_workflow( + "FailingWorkflow", + str, + execution_start_to_close_timeout=timedelta(seconds=30), + ) + + @registry.workflow class UpsertSearchAttributesWorkflow: @workflow.run @@ -313,6 +346,20 @@ async def test_signal_and_query(env: TestWorkflowEnvironment): assert env.get_workflow_result(str, execution.workflow_id) == "approved:yes" +@pytest.mark.asyncio +async def test_failing_signal_handler_surfaces_original_error( + env: TestWorkflowEnvironment, +): + execution = await env.client.start_workflow("SignalWorkflow", task_list="tl") + + await env.client.signal_workflow(execution.workflow_id, "", "reject", "nope") + + assert env.is_workflow_completed(execution.workflow_id) + error = env.get_workflow_error(execution.workflow_id) + assert isinstance(error, ValueError) + assert str(error) == "rejected: nope" + + @pytest.mark.asyncio async def test_signal_external_workflow(env: TestWorkflowEnvironment): await env.client.start_workflow( @@ -409,6 +456,45 @@ async def test_failing_workflow(env: TestWorkflowEnvironment): env.get_workflow_result() +@pytest.mark.asyncio +async def test_activity_application_error_surfaces_as_activity_error( + env: TestWorkflowEnvironment, +): + def fail(name: str) -> str: + raise ApplicationError("not-found", {"name": name}) + + env.on_activity("greet", fn=fail) + await env.client.start_workflow("CatchActivityErrorWorkflow", task_list="tl") + assert env.get_workflow_result(str) == "greet:not-found:{'name': 'dave'}" + + +@pytest.mark.asyncio +async def test_activity_exception_surfaces_with_application_error_cause( + env: TestWorkflowEnvironment, +): + def fail(name: str) -> str: + raise ValueError("boom") + + env.on_activity("greet", fn=fail) + await env.client.start_workflow("ActivityWorkflow", "erin", task_list="tl") + error = env.get_workflow_error() + assert isinstance(error, ActivityError) + assert isinstance(error.cause, ApplicationError) + assert error.cause.reason == "ValueError" + assert "boom" in str(error.cause) + + +@pytest.mark.asyncio +async def test_child_workflow_failure_surfaces_as_child_workflow_error( + env: TestWorkflowEnvironment, +): + await env.client.start_workflow("FailingChildParentWorkflow", task_list="tl") + error = env.get_workflow_error() + assert isinstance(error, ChildWorkflowError) + assert isinstance(error.cause, ApplicationError) + assert error.cause.reason == "ValueError" + + @pytest.mark.asyncio async def test_context_manager_closes(): with TestWorkflowEnvironment(registry) as env: diff --git a/tests/canary/cancellation.py b/tests/canary/cancellation.py index e4301fa..98be6ee 100644 --- a/tests/canary/cancellation.py +++ b/tests/canary/cancellation.py @@ -2,7 +2,6 @@ from datetime import timedelta from cadence import Registry, workflow -from cadence.error import ChildWorkflowExecutionCanceled from cadence.workflow import WorkflowContext from tests.canary.constants import ( @@ -39,6 +38,6 @@ async def run(self) -> None: try: await handle - except (asyncio.CancelledError, ChildWorkflowExecutionCanceled): + except asyncio.CancelledError: return raise RuntimeError("child workflow completed without being cancelled") diff --git a/tests/canary/timeout.py b/tests/canary/timeout.py index 45d8b70..06fa708 100644 --- a/tests/canary/timeout.py +++ b/tests/canary/timeout.py @@ -2,7 +2,7 @@ from datetime import timedelta from cadence import Registry, workflow -from cadence.error import ActivityFailure +from cadence.error import ActivityError, CadenceTimeoutError from tests.canary.constants import ( ACTIVITY_TYPE_TIMEOUT, @@ -29,8 +29,8 @@ async def run(self) -> None: schedule_to_start_timeout=timedelta(seconds=1), start_to_close_timeout=timedelta(seconds=1), ) - except ActivityFailure as error: - if not str(error).startswith("TIMEOUT_TYPE_"): + except ActivityError as error: + if not isinstance(error.cause, CadenceTimeoutError): raise else: raise RuntimeError("expected activity to time out") diff --git a/tests/integration_tests/workflow/test_failures.py b/tests/integration_tests/workflow/test_failures.py new file mode 100644 index 0000000..37f2c5c --- /dev/null +++ b/tests/integration_tests/workflow/test_failures.py @@ -0,0 +1,223 @@ +from datetime import timedelta + +from cadence import Registry, activity, workflow +from cadence.api.v1.common_pb2 import WorkflowExecution +from cadence.api.v1.history_pb2 import EventFilterType, HistoryEvent +from cadence.api.v1.service_workflow_pb2 import GetWorkflowExecutionHistoryRequest +from cadence.client import Client +from cadence.workflow import RetryPolicy +from cadence.error import ( + ActivityError, + ApplicationError, + ChildWorkflowError, + FailureCategory, +) +from tests.integration_tests.helper import CadenceHelper, DOMAIN_NAME + +registry = Registry() + +NO_RETRY: RetryPolicy = { + "maximum_attempts": 1, + "initial_interval": timedelta(seconds=1), + "backoff_coefficient": 1.0, +} + + +@registry.activity() +async def application_failure(code: int) -> None: + raise ApplicationError( + "lookup-failed", + {"code": code}, + "not found", + failure_category=FailureCategory.FATAL, + ) + + +@registry.activity() +async def retry_with_delay() -> int: + attempt = activity.info().attempt + if attempt == 0: + raise ApplicationError("try-again", next_retry_delay=timedelta(seconds=1)) + return attempt + + +@registry.workflow() +class CatchActivityApplicationErrorWorkflow: + @workflow.run + async def run(self, code: int) -> list: + try: + await application_failure.with_options( + schedule_to_close_timeout=timedelta(seconds=10), + retry_policy=NO_RETRY, + ).execute(code) + except ActivityError as e: + assert isinstance(e.cause, ApplicationError) + return [e.cause.reason, *e.cause.details(dict, str)] + raise RuntimeError("expected activity to fail") + + +@registry.workflow() +class ReraiseActivityApplicationErrorWorkflow: + @workflow.run + async def run(self, code: int) -> None: + try: + await application_failure.with_options( + schedule_to_close_timeout=timedelta(seconds=10), + retry_policy=NO_RETRY, + ).execute(code) + except ActivityError as e: + assert e.cause is not None + raise e.cause + + +@registry.workflow() +class ApplicationErrorWorkflow: + @workflow.run + async def run(self, code: int) -> None: + raise ApplicationError("workflow-failed", {"code": code}) + + +@registry.workflow() +class CatchChildApplicationErrorWorkflow: + @workflow.run + async def run(self, code: int) -> list: + try: + await workflow.execute_child_workflow( + "ApplicationErrorWorkflow", + type(None), + code, + execution_start_to_close_timeout=timedelta(seconds=10), + ) + except ChildWorkflowError as e: + assert isinstance(e.cause, ApplicationError) + return [e.cause.reason, *e.cause.details(dict)] + raise RuntimeError("expected child workflow to fail") + + +@registry.workflow() +class NextRetryDelayWorkflow: + @workflow.run + async def run(self) -> int: + return await retry_with_delay.with_options( + schedule_to_close_timeout=timedelta(seconds=60), + start_to_close_timeout=timedelta(seconds=5), + retry_policy={ + "initial_interval": timedelta(seconds=30), + "backoff_coefficient": 1.0, + "maximum_attempts": 3, + }, + ).execute() + + +async def test_activity_application_error_details_round_trip(helper: CadenceHelper): + async with helper.worker(registry) as worker: + execution = await worker.client.start_workflow( + "CatchActivityApplicationErrorWorkflow", + 7, + task_list=worker.task_list, + execution_start_to_close_timeout=timedelta(seconds=30), + ) + + close_event = await _close_event(worker.client, execution) + result = close_event.workflow_execution_completed_event_attributes.result + assert result.data == b'["lookup-failed",{"code":7},"not found"]' + + failed = [ + e.activity_task_failed_event_attributes + for e in await _history(worker.client, execution) + if e.HasField("activity_task_failed_event_attributes") + ] + assert len(failed) == 1 + assert failed[0].failure.reason == "lookup-failed" + assert failed[0].failure.details == b'{"code":7} "not found"' + + +async def test_reraised_activity_application_error_fails_workflow( + helper: CadenceHelper, +): + async with helper.worker(registry) as worker: + execution = await worker.client.start_workflow( + "ReraiseActivityApplicationErrorWorkflow", + 7, + task_list=worker.task_list, + execution_start_to_close_timeout=timedelta(seconds=30), + ) + + close_event = await _close_event(worker.client, execution) + failure = close_event.workflow_execution_failed_event_attributes.failure + assert failure.reason == "lookup-failed" + assert failure.details == b'{"code":7} "not found"' + + +async def test_workflow_application_error_reason_and_details(helper: CadenceHelper): + async with helper.worker(registry) as worker: + execution = await worker.client.start_workflow( + "ApplicationErrorWorkflow", + 3, + task_list=worker.task_list, + execution_start_to_close_timeout=timedelta(seconds=10), + ) + + close_event = await _close_event(worker.client, execution) + failure = close_event.workflow_execution_failed_event_attributes.failure + assert failure.reason == "workflow-failed" + assert failure.details == b'{"code":3}' + + +async def test_child_workflow_application_error(helper: CadenceHelper): + async with helper.worker(registry) as worker: + execution = await worker.client.start_workflow( + "CatchChildApplicationErrorWorkflow", + 5, + task_list=worker.task_list, + execution_start_to_close_timeout=timedelta(seconds=30), + ) + + close_event = await _close_event(worker.client, execution) + result = close_event.workflow_execution_completed_event_attributes.result + assert result.data == b'["workflow-failed",{"code":5}]' + + +async def test_next_retry_delay_overrides_retry_policy_interval( + helper: CadenceHelper, +): + """The retry policy waits 30s between attempts, but the activity asks for 1s.""" + async with helper.worker(registry) as worker: + execution = await worker.client.start_workflow( + "NextRetryDelayWorkflow", + task_list=worker.task_list, + execution_start_to_close_timeout=timedelta(seconds=60), + ) + + close_event = await _close_event(worker.client, execution) + result = close_event.workflow_execution_completed_event_attributes.result + assert result.data == b"1" + + events = await _history(worker.client, execution) + elapsed = events[-1].event_time.ToDatetime() - events[0].event_time.ToDatetime() + assert elapsed < timedelta(seconds=20) + + +async def _close_event(client: Client, execution: WorkflowExecution) -> HistoryEvent: + response = await client.workflow_stub.GetWorkflowExecutionHistory( + GetWorkflowExecutionHistoryRequest( + domain=DOMAIN_NAME, + workflow_execution=execution, + wait_for_new_event=True, + history_event_filter_type=EventFilterType.EVENT_FILTER_TYPE_CLOSE_EVENT, + skip_archival=True, + ) + ) + event: HistoryEvent = response.history.events[-1] + return event + + +async def _history(client: Client, execution: WorkflowExecution) -> list[HistoryEvent]: + response = await client.workflow_stub.GetWorkflowExecutionHistory( + GetWorkflowExecutionHistoryRequest( + domain=DOMAIN_NAME, + workflow_execution=execution, + skip_archival=True, + ) + ) + return list(response.history.events)