Ensure return value serialization errors are marked as task failures

Author: RealOrangeOne

django_tasks/backends/rq.py

@@ -10,6 +10,7 @@
 from django.db import transaction
 from django.utils.functional import cached_property
 from redis.client import Redis
+from rq.defaults import UNSERIALIZABLE_RETURN_VALUE_PAYLOAD
 from rq.exceptions import NoSuchJobError
 from rq.job import Callback, JobStatus
 from rq.job import Job as BaseJob
@@ -116,9 +117,6 @@ def task_result(self) -> TaskResult:
 
         exception_classes = self.meta.get("_django_tasks_exceptions", []).copy()
 
-        if self.worker_name and task_result.status == ResultStatus.RUNNING:
-            task_result.worker_ids.append(self.worker_name)
-
         rq_results = self.results()
 
         for rq_result in rq_results:
@@ -135,11 +133,30 @@ def task_result(self) -> TaskResult:
                     )
                 )
 
+        if self.worker_name and task_result.status == ResultStatus.RUNNING:
+            task_result.worker_ids.append(self.worker_name)
+
         if rq_results:
-            object.__setattr__(task_result, "_return_value", rq_results[0].return_value)
             object.__setattr__(
-                task_result, "last_attempted_at", rq_results[0].created_at
+                task_result, "_return_value", rq_results[-1].return_value
             )
+            object.__setattr__(
+                task_result, "last_attempted_at", rq_results[-1].created_at
+            )
+
+        # If the return value couldn't be serialized, a specific string is saved instead.
+        if task_result._return_value == UNSERIALIZABLE_RETURN_VALUE_PAYLOAD:
+            # In these cases, the task should be marked as failed instead
+            object.__setattr__(task_result, "status", ResultStatus.FAILED)
+
+            task_result.errors.append(
+                TaskError(
+                    exception_class_path=get_module_path(Exception),
+                    traceback=task_result._return_value,  # The traceback isn't stored, so this is the best we can do
+                )
+            )
+
+            object.__setattr__(task_result, "_return_value", None)
 
         return task_result
 

tests/tasks.py

@@ -44,6 +44,12 @@ def complex_exception() -> None:
     raise ValueError(ValueError("This task failed"))
 
 
+@task()
+def complex_return_value() -> Any:
+    # Return something which isn't JSON serializable nor picklable
+    return lambda: True
+
+
 @task()
 def exit_task() -> None:
     exit(1)

tests/tests/test_database_backend.py

@@ -636,6 +636,24 @@ def test_complex_exception(self) -> None:
 
         self.assertEqual(DBTaskResult.objects.ready().count(), 0)
 
+    def test_complex_return_value(self) -> None:
+        result = test_tasks.complex_return_value.enqueue()
+
+        self.run_worker()
+
+        result.refresh()
+
+        self.assertEqual(result.status, ResultStatus.FAILED)
+        self.assertIsNotNone(result.started_at)
+        self.assertIsNotNone(result.last_attempted_at)
+        self.assertIsNotNone(result.finished_at)
+        self.assertGreaterEqual(result.started_at, result.enqueued_at)  # type:ignore[arg-type,misc]
+        self.assertGreaterEqual(result.finished_at, result.started_at)  # type:ignore[arg-type,misc]
+
+        self.assertIsNone(result._return_value)
+        self.assertEqual(result.errors[0].exception_class, TypeError)
+        self.assertIn("is not JSON serializable", result.errors[0].traceback)
+
     def test_doesnt_process_different_backend(self) -> None:
         result = test_tasks.failing_task_value_error.enqueue()
 

tests/tests/test_immediate_backend.py

@@ -137,6 +137,21 @@ def test_complex_exception(self) -> None:
         self.assertEqual(result.args, [])
         self.assertEqual(result.kwargs, {})
 
+    def test_complex_return_value(self) -> None:
+        with self.assertLogs("django_tasks", level="ERROR"):
+            result = test_tasks.complex_return_value.enqueue()
+
+        self.assertEqual(result.status, ResultStatus.FAILED)
+        self.assertIsNotNone(result.started_at)
+        self.assertIsNotNone(result.last_attempted_at)
+        self.assertIsNotNone(result.finished_at)
+        self.assertGreaterEqual(result.started_at, result.enqueued_at)  # type:ignore[arg-type,misc]
+        self.assertGreaterEqual(result.finished_at, result.started_at)  # type:ignore[arg-type,misc]
+
+        self.assertIsNone(result._return_value)
+        self.assertEqual(result.errors[0].exception_class, TypeError)
+        self.assertIn("is not JSON serializable", result.errors[0].traceback)
+
     def test_result(self) -> None:
         result = default_task_backend.enqueue(
             test_tasks.calculate_meaning_of_life, [], {}

tests/tests/test_rq_backend.py

@@ -11,6 +11,7 @@
 from django.test import TransactionTestCase, modify_settings, override_settings
 from django.urls import reverse
 from fakeredis import FakeRedis, FakeStrictRedis
+from rq.defaults import UNSERIALIZABLE_RETURN_VALUE_PAYLOAD
 from rq.timeouts import TimerDeathPenalty
 
 from django_tasks import ResultStatus, Task, default_task_backend, tasks
@@ -199,6 +200,25 @@ def test_complex_exception(self) -> None:
         self.assertEqual(result.kwargs, {})
         self.assertEqual(result.attempts, 1)
 
+    def test_complex_return_value(self) -> None:
+        result = test_tasks.complex_return_value.enqueue()
+
+        with self.assertLogs("django_tasks", "DEBUG"):
+            self.run_worker()
+
+        result.refresh()
+
+        self.assertEqual(result.status, ResultStatus.FAILED)
+        self.assertIsNotNone(result.started_at)
+        self.assertIsNotNone(result.last_attempted_at)
+        self.assertIsNotNone(result.finished_at)
+        self.assertGreaterEqual(result.started_at, result.enqueued_at)  # type:ignore[arg-type,misc]
+        self.assertGreaterEqual(result.finished_at, result.started_at)  # type:ignore[arg-type,misc]
+
+        self.assertIsNone(result._return_value)
+        self.assertEqual(result.errors[0].exception_class, Exception)
+        self.assertIn(UNSERIALIZABLE_RETURN_VALUE_PAYLOAD, result.errors[0].traceback)
+
     def test_get_result(self) -> None:
         result = default_task_backend.enqueue(test_tasks.noop_task, [], {})