Minor exception-related updates (#75)

Author: cretz

temporalio/exceptions.py

@@ -2,7 +2,7 @@
 
 import traceback
 from enum import IntEnum
-from typing import Any, Awaitable, Callable, Iterable, Optional
+from typing import Any, Awaitable, Callable, Iterable, Optional, Tuple
 
 import temporalio.api.common.v1
 import temporalio.api.enums.v1
@@ -39,10 +39,13 @@ def __init__(
         self,
         message: str,
         *,
-        failure: Optional[temporalio.api.failure.v1.Failure] = None
+        failure: Optional[temporalio.api.failure.v1.Failure] = None,
+        exc_args: Optional[Tuple] = None,
     ) -> None:
         """Initialize a failure error."""
-        super().__init__(message)
+        if exc_args is None:
+            exc_args = (message,)
+        super().__init__(*exc_args)
         self._message = message
         self._failure = failure
 
@@ -65,10 +68,14 @@ def __init__(
         message: str,
         *details: Any,
         type: Optional[str] = None,
-        non_retryable: bool = False
+        non_retryable: bool = False,
     ) -> None:
         """Initialize an application error."""
-        super().__init__(message)
+        super().__init__(
+            message,
+            # If there is a type, prepend it to the message on the string repr
+            exc_args=(message if not type else f"{type}: {message}",),
+        )
         self._details = details
         self._type = type
         self._non_retryable = non_retryable
@@ -140,7 +147,7 @@ def __init__(
         message: str,
         *,
         type: Optional[TimeoutType],
-        last_heartbeat_details: Iterable[Any]
+        last_heartbeat_details: Iterable[Any],
     ) -> None:
         """Initialize a timeout error."""
         super().__init__(message)
@@ -206,7 +213,7 @@ def __init__(
         identity: str,
         activity_type: str,
         activity_id: str,
-        retry_state: Optional[RetryState]
+        retry_state: Optional[RetryState],
     ) -> None:
         """Initialize an activity error."""
         super().__init__(message)
@@ -261,7 +268,7 @@ def __init__(
         workflow_type: str,
         initiated_event_id: int,
         started_event_id: int,
-        retry_state: Optional[RetryState]
+        retry_state: Optional[RetryState],
     ) -> None:
         """Initialize a child workflow error."""
         super().__init__(message)
@@ -404,7 +411,7 @@ def apply_error_to_failure(
 
     # Set message, stack, and cause. Obtaining cause follows rules from
     # https://docs.python.org/3/library/exceptions.html#exception-context
-    failure.message = str(error)
+    failure.message = error.message
     if error.__traceback__:
         failure.stack_trace = "\n".join(traceback.format_tb(error.__traceback__))
     if error.__cause__:
@@ -490,6 +497,7 @@ def apply_exception_to_failure(
             str(exception), type=exception.__class__.__name__
         )
         failure_error.__traceback__ = exception.__traceback__
+        failure_error.__cause__ = exception.__cause__
         apply_error_to_failure(failure_error, converter, failure)
 
 

temporalio/worker/activity.py

@@ -418,7 +418,7 @@ async def _run_activity(
                     and running_activity.cancelled_due_to_heartbeat_error
                 ):
                     err = running_activity.cancelled_due_to_heartbeat_error
-                    temporalio.activity.logger.debug(
+                    temporalio.activity.logger.warning(
                         f"Completing as failure during heartbeat with error of type {type(err)}: {err}",
                     )
                     await temporalio.exceptions.encode_exception_to_failure(
@@ -438,8 +438,8 @@ async def _run_activity(
                         completion.result.cancelled.failure,
                     )
                 else:
-                    temporalio.activity.logger.debug(
-                        "Completing as failed", exc_info=True
+                    temporalio.activity.logger.warning(
+                        "Completing activity as failed", exc_info=True
                     )
                     await temporalio.exceptions.encode_exception_to_failure(
                         err,

temporalio/worker/workflow_instance.py

@@ -265,7 +265,10 @@ def activate(
                 # Run one iteration of the loop
                 self._run_once()
         except Exception as err:
-            logger.exception(f"Failed activation on workflow with run ID {act.run_id}")
+            logger.warning(
+                f"Failed activation on workflow {self._info.workflow_type} with ID {self._info.workflow_id} and run ID {self._info.run_id}",
+                exc_info=True,
+            )
             # Set completion failure
             self._current_completion.failed.failure.SetInParent()
             try:

tests/test_exceptions.py

@@ -0,0 +1,71 @@
+import logging
+import logging.handlers
+import traceback
+from typing import Optional
+
+import temporalio.converter
+from temporalio.api.failure.v1 import Failure
+from temporalio.exceptions import (
+    ApplicationError,
+    FailureError,
+    apply_exception_to_failure,
+    failure_to_error,
+)
+
+
+# This is an example of appending the stack to every Temporal failure error
+def append_temporal_stack(exc: Optional[BaseException]) -> None:
+    while exc:
+        # Only append if it doesn't appear already there
+        if (
+            isinstance(exc, FailureError)
+            and exc.failure
+            and exc.failure.stack_trace
+            and len(exc.args) == 1
+            and "\nStack:\n" not in exc.args[0]
+        ):
+            exc.args = (f"{exc}\nStack:\n{exc.failure.stack_trace.rstrip()}",)
+        exc = exc.__cause__
+
+
+def test_exception_format():
+    # Cause a nested exception
+    actual_err: Exception
+    try:
+        try:
+            raise ValueError("error1")
+        except Exception as err:
+            raise RuntimeError("error2") from err
+    except Exception as err:
+        actual_err = err
+    assert actual_err
+
+    # Convert to failure and back
+    failure = Failure()
+    apply_exception_to_failure(
+        actual_err, temporalio.converter.default().payload_converter, failure
+    )
+    failure_error = failure_to_error(
+        failure, temporalio.converter.default().payload_converter
+    )
+    # Confirm type is prepended
+    assert isinstance(failure_error, ApplicationError)
+    assert "RuntimeError: error2" == str(failure_error)
+    assert isinstance(failure_error.cause, ApplicationError)
+    assert "ValueError: error1" == str(failure_error.cause)
+
+    # Append the stack and format the exception and check the output
+    append_temporal_stack(failure_error)
+    output = "".join(
+        traceback.format_exception(
+            type(failure_error), failure_error, failure_error.__traceback__
+        )
+    )
+    assert "temporalio.exceptions.ApplicationError: ValueError: error1" in output
+    assert "temporalio.exceptions.ApplicationError: RuntimeError: error" in output
+    assert output.count("\nStack:\n") == 2
+
+    # This shows how it might look for those with debugging on
+    logging.getLogger(__name__).debug(
+        "Showing appended exception", exc_info=failure_error
+    )

tests/worker/test_activity.py

@@ -172,7 +172,7 @@ async def raise_error():
 
     with pytest.raises(WorkflowFailureError) as err:
         await _execute_workflow_with_activity(client, worker, raise_error)
-    assert str(assert_activity_application_error(err.value)) == "oh no!"
+    assert str(assert_activity_application_error(err.value)) == "RuntimeError: oh no!"
 
 
 @activity.defn
@@ -189,7 +189,7 @@ async def test_sync_activity_process_failure(client: Client, worker: ExternalWor
                 picklable_activity_failure,
                 worker_config={"activity_executor": executor},
             )
-    assert str(assert_activity_application_error(err.value)) == "oh no!"
+    assert str(assert_activity_application_error(err.value)) == "RuntimeError: oh no!"
 
 
 async def test_activity_bad_params(client: Client, worker: ExternalWorker):
@@ -328,7 +328,7 @@ async def say_hello(name: str) -> str:
                 task_queue=worker.task_queue,
             )
     assert str(assert_activity_application_error(err.value)) == (
-        "Activity function wrong_activity is not registered on this worker, "
+        "NotFoundError: Activity function wrong_activity is not registered on this worker, "
         "available activities: say_hello"
     )
 
@@ -565,7 +565,10 @@ async def some_activity():
             retry_max_attempts=100,
             non_retryable_error_types=["Cannot retry me"],
         )
-    assert str(assert_activity_application_error(err.value)) == "Do not retry me"
+    assert (
+        str(assert_activity_application_error(err.value))
+        == "Cannot retry me: Do not retry me"
+    )
 
 
 async def test_activity_logging(client: Client, worker: ExternalWorker):