changed how error is transferred

Author: Andrei Neagu

packages/service-library/src/servicelib/long_running_tasks/_lrt_client.py

@@ -5,13 +5,12 @@
 from models_library.rabbitmq_basic_types import RPCMethodName
 from pydantic import PositiveInt, TypeAdapter
 
-from ..logging_errors import create_troubleshootting_log_kwargs
 from ..logging_utils import log_decorator
 from ..rabbitmq._client_rpc import RabbitMQRPCClient
 from ._rabbit_namespace import get_rabbit_namespace
 from ._serialization import string_to_object
+from .errors import RPCTransferrableTaskError
 from .models import (
-    ErrorResponse,
     LRTNamespace,
     RegisteredTaskName,
     TaskBase,
@@ -97,34 +96,19 @@ async def get_task_result(
     task_context: TaskContext,
     task_id: TaskId,
 ) -> Any:
-    serialized_result = await rabbitmq_rpc_client.request(
-        get_rabbit_namespace(namespace),
-        TypeAdapter(RPCMethodName).validate_python("get_task_result"),
-        task_context=task_context,
-        task_id=task_id,
-        timeout_s=_RPC_TIMEOUT_SHORT_REQUESTS,
-    )
-    assert isinstance(serialized_result, ErrorResponse | str)  # nosec
-    if isinstance(serialized_result, ErrorResponse):
-        error = string_to_object(serialized_result.str_error_object)
-        _logger.info(
-            **create_troubleshootting_log_kwargs(
-                f"Task '{task_id}' raised the following error:\n{serialized_result.str_traceback}",
-                error=error,
-                error_context={
-                    "task_id": task_id,
-                    "namespace": namespace,
-                    "task_context": task_context,
-                },
-                tip=(
-                    f"The caller of this function should handle the exception. "
-                    f"To figure out where it was running check {namespace=}"
-                ),
-            )
+    try:
+        serialized_result = await rabbitmq_rpc_client.request(
+            get_rabbit_namespace(namespace),
+            TypeAdapter(RPCMethodName).validate_python("get_task_result"),
+            task_context=task_context,
+            task_id=task_id,
+            timeout_s=_RPC_TIMEOUT_SHORT_REQUESTS,
         )
-        raise error
-
-    return string_to_object(serialized_result)
+        assert isinstance(serialized_result, str)  # nosec
+        return string_to_object(serialized_result)
+    except RPCTransferrableTaskError as e:
+        decoded_error = string_to_object(f"{e}")
+        raise decoded_error from None
 
 
 @log_decorator(_logger, level=logging.DEBUG)

packages/service-library/src/servicelib/long_running_tasks/_lrt_server.py

@@ -2,12 +2,9 @@
 from contextlib import suppress
 from typing import TYPE_CHECKING, Any
 
-from ..logging_errors import create_troubleshootting_log_kwargs
 from ..rabbitmq import RPCRouter
-from ._serialization import string_to_object
-from .errors import BaseLongRunningError, TaskNotFoundError
+from .errors import BaseLongRunningError, RPCTransferrableTaskError, TaskNotFoundError
 from .models import (
-    ErrorResponse,
     RegisteredTaskName,
     TaskBase,
     TaskContext,
@@ -66,43 +63,19 @@ async def get_task_status(
     )
 
 
-@router.expose(reraise_if_error_type=(BaseLongRunningError,))
+@router.expose(reraise_if_error_type=(BaseLongRunningError, RPCTransferrableTaskError))
 async def get_task_result(
     long_running_manager: "BaseLongRunningManager",
     *,
     task_context: TaskContext,
     task_id: TaskId,
-) -> ErrorResponse | str:
+) -> str:
     try:
         result_field = await long_running_manager.tasks_manager.get_task_result(
             task_id, with_task_context=task_context
         )
-        if result_field.error_response is not None:
-            task_raised_error_traceback = result_field.error_response.str_traceback
-            task_raised_error = string_to_object(
-                result_field.error_response.str_error_object
-            )
-            _logger.info(
-                **create_troubleshootting_log_kwargs(
-                    f"Execution of {task_id=} finished with error:\n{task_raised_error_traceback}",
-                    error=task_raised_error,
-                    error_context={
-                        "task_id": task_id,
-                        "task_context": task_context,
-                        "namespace": long_running_manager.lrt_namespace,
-                    },
-                    tip="This exception is logged for debugging purposes, the client side will handle it",
-                )
-            )
-            allowed_errors = (
-                await long_running_manager.tasks_manager.get_allowed_errors(
-                    task_id, with_task_context=task_context
-                )
-            )
-            if type(task_raised_error) in allowed_errors:
-                return result_field.error_response
-
-            raise task_raised_error
+        if result_field.str_error is not None:
+            raise RPCTransferrableTaskError(result_field.str_error)
 
         if result_field.str_result is not None:
             return result_field.str_result

packages/service-library/src/servicelib/long_running_tasks/errors.py

@@ -44,3 +44,10 @@ class GenericClientError(BaseLongRunningError):
     msg_template: str = (
         "Unexpected error while '{action}' for '{task_id}': status={status} body={body}"
     )
+
+
+class RPCTransferrableTaskError(Exception):
+    """
+    The message contains the task's exception serialized as string.
+    Decode it and raise to obtain the task's original exception.
+    """

packages/service-library/src/servicelib/long_running_tasks/models.py

@@ -33,18 +33,13 @@
 RegisteredTaskName: TypeAlias = str
 
 
-class ErrorResponse(BaseModel):
-    str_error_object: str
-    str_traceback: str
-
-
 class ResultField(BaseModel):
     str_result: str | None = None
-    error_response: ErrorResponse | None = None
+    str_error: str | None = None
 
     @model_validator(mode="after")
     def validate_mutually_exclusive(self) -> "ResultField":
-        if self.str_result is not None and self.error_response is not None:
+        if self.str_result is not None and self.str_error is not None:
             msg = "Cannot set both 'result' and 'error' - they are mutually exclusive"
             raise ValueError(msg)
         return self

packages/service-library/src/servicelib/long_running_tasks/task.py

@@ -3,7 +3,6 @@
 import functools
 import inspect
 import logging
-import traceback
 import urllib.parse
 from contextlib import suppress
 from typing import Any, ClassVar, Final, Protocol, TypeAlias
@@ -22,6 +21,7 @@
 )
 
 from ..background_task import create_periodic_task
+from ..logging_errors import create_troubleshootting_log_kwargs
 from ..redis import RedisClientSDK, exclusive
 from ._redis_store import RedisStore
 from ._serialization import object_to_string
@@ -33,7 +33,6 @@
     TaskNotRegisteredError,
 )
 from .models import (
-    ErrorResponse,
     LRTNamespace,
     RegisteredTaskName,
     ResultField,
@@ -321,22 +320,30 @@ async def _tasks_monitor(self) -> None:
                 except asyncio.InvalidStateError:
                     # task was not completed try again next time and see if it is done
                     continue
-                except asyncio.CancelledError as e:
+                except asyncio.CancelledError:
                     result_field = ResultField(
-                        error_response=ErrorResponse(
-                            str_error_object=object_to_string(
-                                TaskCancelledError(task_id=task_id)
-                            ),
-                            str_traceback="".join(traceback.format_tb(e.__traceback__)),
-                        )
+                        str_error=object_to_string(TaskCancelledError(task_id=task_id))
                     )
                 except Exception as e:  # pylint:disable=broad-except
-                    result_field = ResultField(
-                        error_response=ErrorResponse(
-                            str_error_object=object_to_string(e),
-                            str_traceback="".join(traceback.format_tb(e.__traceback__)),
-                        )
+                    allowed_errors = TaskRegistry.get_allowed_errors(
+                        task_data.registered_task_name
                     )
+                    if type(e) not in allowed_errors:
+                        _logger.exception(
+                            **create_troubleshootting_log_kwargs(
+                                (
+                                    f"Execution of {task_id=} finished with unexpected error, "
+                                    f"only the following are {allowed_errors=} are permitted"
+                                ),
+                                error=e,
+                                error_context={
+                                    "task_id": task_id,
+                                    "task_data": task_data,
+                                    "namespace": self.lrt_namespace,
+                                },
+                            ),
+                        )
+                    result_field = ResultField(str_error=object_to_string(e))
 
                 # update and store in Redis
                 updates = {"is_done": is_done, "result_field": task_data.result_field}

packages/service-library/tests/long_running_tasks/test_long_running_tasks_task.py

@@ -378,8 +378,8 @@ async def test_get_result_finished_with_error(
     result = await long_running_manager.tasks_manager.get_task_result(
         task_id, with_task_context=empty_context
     )
-    assert result.error_response is not None  # nosec
-    error = string_to_object(result.error_response.str_error_object)
+    assert result.str_error is not None  # nosec
+    error = string_to_object(result.str_error)
     with pytest.raises(_TetingError, match="failing asap"):
         raise error