fix(tasks): give task_failure_handler full ownership of FAILURE path

task_postrun_handler now early-returns for FAILURE states instead of
redundantly writing status and date_done. task_failure_handler now
computes duration_seconds and wait_time_seconds so failed tasks get
complete timing data. This eliminates a wasted .get() + .save() round
trip on every failed task and gives each handler a clean, non-overlapping
responsibility.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/documents/signals/handlers.py

@@ -1226,12 +1226,11 @@ def task_postrun_handler(
     **kwargs,
 ) -> None:
     """
-    Records task completion and result data.
+    Records task completion and result data for non-failure outcomes.
 
-    task_failure also fires when a task raises an exception, and it writes
-    richer structured error data.  To avoid a race where this handler
-    overwrites that data, result_data and result_message are left untouched
-    when the final state is FAILURE. task_failure_handler owns those fields.
+    Skips FAILURE states entirely, since task_failure_handler fires first
+    and fully owns the failure path (status, date_done, duration,
+    result_data, result_message).
 
     https://docs.celeryq.dev/en/stable/userguide/signals.html#task-postrun
     """
@@ -1243,6 +1242,8 @@ def task_postrun_handler(
         close_old_connections()
 
         new_status = _CELERY_STATE_TO_STATUS.get(state, PaperlessTask.Status.FAILURE)
+        if new_status == PaperlessTask.Status.FAILURE:
+            return
 
         now = timezone.now()
         try:
@@ -1265,16 +1266,13 @@ def task_postrun_handler(
             ).total_seconds()
             changed_fields.append("wait_time_seconds")
 
-        # Only write result data for non-failure outcomes; task_failure_handler
-        # owns result_data/result_message for FAILURE states.
-        if new_status != PaperlessTask.Status.FAILURE:
-            if isinstance(retval, dict):
-                task_instance.result_data = retval
-                changed_fields.append("result_data")
-            elif isinstance(retval, str):
-                task_instance.result_message = retval
-                task_instance.result_data = _parse_consume_result(retval)
-                changed_fields.extend(["result_message", "result_data"])
+        if isinstance(retval, dict):
+            task_instance.result_data = retval
+            changed_fields.append("result_data")
+        elif isinstance(retval, str):
+            task_instance.result_message = retval
+            task_instance.result_data = _parse_consume_result(retval)
+            changed_fields.extend(["result_message", "result_data"])
 
         task_instance.save(update_fields=changed_fields)
     except Exception:  # pragma: no cover
@@ -1293,6 +1291,9 @@ def task_failure_handler(
     """
     Records failure details when a task raises an exception.
 
+    Fully owns the FAILURE path. task_postrun_handler skips FAILURE
+    states so there is no overlap.
+
     https://docs.celeryq.dev/en/stable/userguide/signals.html#task-failure
     """
     if task_id is None:  # pragma: no cover
@@ -1310,12 +1311,26 @@ def task_failure_handler(
             tb_str = "".join(_tb.format_tb(traceback))
             result_data["traceback"] = tb_str[:5000]
 
-        PaperlessTask.objects.filter(task_id=task_id).update(
-            status=PaperlessTask.Status.FAILURE,
-            result_data=result_data,
-            result_message=str(exception) if exception else None,
-            date_done=timezone.now(),
-        )
+        now = timezone.now()
+        update_fields: dict = {
+            "status": PaperlessTask.Status.FAILURE,
+            "result_data": result_data,
+            "result_message": str(exception) if exception else None,
+            "date_done": now,
+        }
+
+        task_qs = PaperlessTask.objects.filter(task_id=task_id)
+        task_instance = task_qs.values("date_started", "date_created").first()
+        if task_instance:
+            date_started = task_instance["date_started"]
+            if date_started:
+                update_fields["duration_seconds"] = (now - date_started).total_seconds()
+            date_created = task_instance["date_created"]
+            if date_started and date_created:
+                update_fields["wait_time_seconds"] = (
+                    date_started - date_created
+                ).total_seconds()
+            task_qs.update(**update_fields)
     except Exception:  # pragma: no cover
         logger.exception("Updating PaperlessTask on failure failed")
 

src/documents/tests/test_task_signals.py

@@ -193,13 +193,14 @@ class TestTaskPostrunHandler:
         assert task.duration_seconds is not None
         assert task.wait_time_seconds is not None
 
-    def test_records_failure_state(self):
+    def test_skips_failure_state(self):
+        """postrun skips FAILURE; task_failure_handler owns that path."""
         task = self._started_task()
         from documents.signals.handlers import task_postrun_handler
 
         task_postrun_handler(task_id=task.task_id, retval="some error", state="FAILURE")
         task.refresh_from_db()
-        assert task.status == PaperlessTask.Status.FAILURE
+        assert task.status == PaperlessTask.Status.STARTED
 
     def test_parses_legacy_new_document_string(self):
         task = self._started_task()
@@ -295,6 +296,27 @@ class TestTaskFailureHandler:
         assert "traceback" in task.result_data
         assert len(task.result_data["traceback"]) <= 5000
 
+    def test_computes_duration_and_wait_time(self):
+        from django.utils import timezone
+
+        now = timezone.now()
+        task = PaperlessTaskFactory(
+            task_type=PaperlessTask.TaskType.CONSUME_FILE,
+            status=PaperlessTask.Status.STARTED,
+            date_created=now - timezone.timedelta(seconds=10),
+            date_started=now - timezone.timedelta(seconds=5),
+        )
+        from documents.signals.handlers import task_failure_handler
+
+        task_failure_handler(
+            task_id=task.task_id,
+            exception=ValueError("boom"),
+            traceback=None,
+        )
+        task.refresh_from_db()
+        assert task.duration_seconds == pytest.approx(5.0, abs=1.0)
+        assert task.wait_time_seconds == pytest.approx(5.0, abs=1.0)
+
     def test_ignores_none_task_id(self):
         from documents.signals.handlers import task_failure_handler