handle unpicklable errors

Author: Andrei Neagu

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

@@ -34,6 +34,12 @@ class TaskExceptionError(BaseLongRunningError):
     )
 
 
+class TaskRaisedUnserializableError(BaseLongRunningError):
+    msg_template: str = (
+        "Task {task_id} finished with an unserializable exception: '{exception}'\n{traceback}"
+    )
+
+
 class TaskClientTimeoutError(BaseLongRunningError):
     msg_template: str = (
         "Timed out after {timeout} seconds while awaiting '{task_id}' to complete"

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

@@ -31,6 +31,7 @@
     TaskNotCompletedError,
     TaskNotFoundError,
     TaskNotRegisteredError,
+    TaskRaisedUnserializableError,
 )
 from .models import (
     LRTNamespace,
@@ -345,7 +346,32 @@ async def _tasks_monitor(self) -> None:
                                 },
                             ),
                         )
-                    result_field = ResultField(str_error=dumps(e))
+                    try:
+                        result_field = ResultField(str_error=dumps(e))
+                    except (
+                        Exception  # pylint:disable=broad-except
+                    ) as serialization_error:
+                        _logger.exception(
+                            **create_troubleshootting_log_kwargs(
+                                (
+                                    f"Execution of {task_id=} finished with error "
+                                    f"which could not be serialized"
+                                ),
+                                error=serialization_error,
+                                error_context={
+                                    "task_id": task_id,
+                                    "task_data": task_data,
+                                    "namespace": self.lrt_namespace,
+                                },
+                            ),
+                        )
+                        result_field = ResultField(
+                            str_error=dumps(
+                                TaskRaisedUnserializableError(
+                                    task_id=task_id, exception=serialization_error
+                                )
+                            )
+                        )
 
                 # 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

@@ -25,6 +25,7 @@
     TaskNotCompletedError,
     TaskNotFoundError,
     TaskNotRegisteredError,
+    TaskRaisedUnserializableError,
 )
 from servicelib.long_running_tasks.models import (
     LRTNamespace,
@@ -37,7 +38,7 @@
 from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
 from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
-from tenacity import TryAgain
+from tenacity import TryAgain, retry, stop_after_attempt
 from tenacity.asyncio import AsyncRetrying
 from tenacity.retry import retry_if_exception_type
 from tenacity.stop import stop_after_delay
@@ -77,19 +78,30 @@ async def a_background_task(
 
 
 async def fast_background_task(progress: TaskProgress) -> int:
-    """this task does nothing and returns a constant"""
     return 42
 
 
 async def failing_background_task(progress: TaskProgress):
-    """this task does nothing and returns a constant"""
     msg = "failing asap"
     raise _TetingError(msg)
 
 
+async def failing_unpicklable_background_task(progress: TaskProgress):
+    @retry(
+        stop=stop_after_attempt(2),
+        reraise=False,
+    )
+    async def _innter_fail() -> None:
+        msg = "always fails with retry"
+        raise _TetingError(msg)
+
+    await _innter_fail()
+
+
 TaskRegistry.register(a_background_task)
 TaskRegistry.register(fast_background_task)
 TaskRegistry.register(failing_background_task)
+TaskRegistry.register(failing_unpicklable_background_task)
 
 
 @pytest.fixture
@@ -385,6 +397,33 @@ async def test_get_result_finished_with_error(
         raise error
 
 
+async def test_get_result_finished_with_unpicklable_error(
+    long_running_manager: BaseLongRunningManager, empty_context: TaskContext
+):
+    task_id = await lrt_api.start_task(
+        long_running_manager.rpc_client,
+        long_running_manager.lrt_namespace,
+        failing_unpicklable_background_task.__name__,
+        task_context=empty_context,
+    )
+    # wait for result
+    async for attempt in AsyncRetrying(**_RETRY_PARAMS):
+        with attempt:
+            assert (
+                await long_running_manager.tasks_manager.get_task_status(
+                    task_id, with_task_context=empty_context
+                )
+            ).done
+
+    result = await long_running_manager.tasks_manager.get_task_result(
+        task_id, with_task_context=empty_context
+    )
+    assert result.str_error is not None  # nosec
+    error = loads(result.str_error)
+    with pytest.raises(TaskRaisedUnserializableError, match="cannot pickle"):
+        raise error
+
+
 async def test_cancel_task_from_different_manager(
     rabbit_service: RabbitSettings,
     use_in_memory_redis: RedisSettings,