From 2be6a55ae2f485a7efd7eb945b618f9bd85a8ba9 Mon Sep 17 00:00:00 2001 From: Nathanael Mortensen Date: Thu, 15 Oct 2026 12:28:26 -0700 Subject: [PATCH] feat!: Standardize Error Handling and Support Custom Error Reasons This is a breaking change! Standardize the exception hierarchy, introducing CadenceError as the root of all Errors and FailureError as the root of any Failure expected in Workflow code. We call it FailureError rather than WorkflowError as ApplicationError is raised by Activity implementations. Add ApplicationError. This allows users to explicitly specify the error reason, details, and any FailureOptions from a Workflow or an Activity. If a Workflow or Activity raises any other type of Error then the reason will be the name of the error type and the details will be a serialized representation of the error. We don't attempt to make this representation of the error round-trippable. It's purely for manual inspection and debugging, not to propagate stacktraces through a Workflow. Within Workflows we introduce a consistent hierarchy for receiving exceptions from Activities or Child Workflows. For Activities: - If cancelled, we raise asyncio.CancelledError - Else, we raise ActivityError. The reason for the error is the backing `__cause__`, which we also expose as a property: - ApplicationError if the activity failed. This contains the reason and details. - CadenceTimeoutError if the activity timed out. For Child Workflows: - If cancelled by the Workflow, we raise asyncio.CancelledError - If we failed to start the Child Workflow, we raise StartChildWorkflowExecutionError (extends ChildWorkflowError) - Else, we raise ChildWorkflowError. Again the reason is the backing cause: - ApplicationError if the child workflow failed. - WorkflowCancelledError if the child workflow was cancelled externally - CadenceTimeoutError if the child workflow timed out. - TerminatedError if the child worfklow was terminated. Distinguishing between cancellation from within the Workflow and external Cancellation is important because asyncio.CancelledError extends BaseException instead of Exception. It's unlikely that Workflow code expects a non-Exception type if they're not the one cancelling the Workflow. This aligns with the Java SDK. SignalFailure is removed, it was an implementation detail that was never exposed to users. Usage was replaced with FatalDecisionError. The breaking changes are: - Removed most subtypes of ChildWorkflowError to align it with Activities. - Renamed all errors to end with Error: - ActivityFailure -> ActivityError - WorkflowFailure -> FailureError - SignalExternalWorkflowFailed -> SignalExternalWorkflowError - StartChildWorkflowExecutionError -> StartChildWorkflowExecutionError - Renamed StartChildWorkflowExecutionFailedError.cause -> failure_cause to avoid confusion with Error `cause`. --- .../_internal/activity/_activity_executor.py | 26 +- cadence/_internal/failure.py | 288 ++++++++++++++++ .../statemachine/activity_state_machine.py | 20 +- .../child_workflow_execution_state_machine.py | 62 ++-- .../workflow/statemachine/decision_manager.py | 16 +- .../signal_external_workflow_state_machine.py | 7 +- cadence/_internal/workflow/workflow_engine.py | 32 +- .../_internal/workflow/workflow_instance.py | 12 +- cadence/error.py | 317 +++++++++++++++--- cadence/testing/_workflow_environment.py | 31 +- .../activity/test_activity_executor.py | 69 +++- tests/cadence/_internal/test_failure.py | 129 +++++++ .../test_activity_state_machine.py | 121 +++++-- ..._child_workflow_execution_state_machine.py | 136 +++++++- .../statemachine/test_decision_manager.py | 13 +- ..._signal_external_workflow_state_machine.py | 18 +- .../workflow/test_context_child_workflow.py | 14 +- .../workflow/test_signal_handling.py | 8 +- .../workflow/test_workflow_engine.py | 135 +++++++- tests/cadence/test_context.py | 5 +- .../testing/test_workflow_environment.py | 86 +++++ tests/canary/cancellation.py | 3 +- tests/canary/timeout.py | 6 +- .../workflow/test_failures.py | 223 ++++++++++++ 24 files changed, 1559 insertions(+), 218 deletions(-) create mode 100644 cadence/_internal/failure.py create mode 100644 tests/cadence/_internal/test_failure.py create mode 100644 tests/integration_tests/workflow/test_failures.py diff --git a/cadence/_internal/activity/_activity_executor.py b/cadence/_internal/activity/_activity_executor.py index e2139261..d522f367 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 00000000..73b31fde --- /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 684b61a3..2fd0d11b 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 2ba3a762..6e696c6f 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 5c8b6d15..053430de 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 1ebee1e8..90735748 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 1b5d5ef6..bec0eb4a 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 f73fcfe6..fa73e82e 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 f823ebd9..cbfed450 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 47b00f32..cad3d755 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 74805d7c..927717bd 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 00000000..0e95818b --- /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 95905581..bd822444 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 d02b6f8f..c38d7435 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 6c16d446..ca6e8f2d 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 0f4e0925..edd8d0d5 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 f0f98ed2..7dda4473 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 5baa145a..3f9aece1 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 46d2862a..3e758839 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 1a951a3a..1d2b89f8 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 9b295603..2a39b5c9 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 e4301fa3..98be6ee4 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 45d8b70e..06fa708e 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 00000000..37f2c5c3 --- /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)