Fix: Handle Celery mail task chord errors (#13936)

* Fix: mail rule loops forever when all attachments are duplicates

When every attachment in a mail is rejected as a duplicate, the chord's
header tasks all fail. Celery's default task_allow_error_cb_on_chord_header
skips the error callback in that case, so no ProcessedMail row is ever
created, and the same mail is refetched and reprocessed on every poll for
as long as it stays in the rule's maximum_age window.

* Minor simplifications and cleanup

src/paperless/settings/__init__.py

@@ -705,6 +705,12 @@ CELERY_BROKER_TRANSPORT_OPTIONS = {
 CELERY_TASK_TRACK_STARTED = True
 CELERY_TASK_TIME_LIMIT: Final[int] = get_int_from_env("PAPERLESS_WORKER_TIMEOUT", 1800)
 
+# https://docs.celeryq.dev/en/stable/userguide/configuration.html#std-setting-task_allow_error_cb_on_chord_header
+# Without this, a failing chord header never triggers the errback, so a mail
+# whose attachments all fail is never recorded and is re-fetched forever.
+# The errback runs once per failed header task, so it must be idempotent.
+CELERY_TASK_ALLOW_ERROR_CB_ON_CHORD_HEADER = True
+
 CELERY_CACHE_BACKEND = "default"
 
 # https://docs.celeryq.dev/en/stable/userguide/configuration.html#task-serializer

src/paperless_mail/mail.py

@@ -334,18 +334,24 @@ def error_callback(
     """
     A shared task that is called whenever something goes wrong during
     consumption of a file. See queue_consumption_tasks.
+
+    With CELERY_TASK_ALLOW_ERROR_CB_ON_CHORD_HEADER enabled this runs once per
+    failed header task, not once per chord, so it must be idempotent.
     """
     rule = MailRule.objects.get(pk=rule_id)
+    received = make_aware(message_date) if is_naive(message_date) else message_date
 
-    ProcessedMail.objects.create(
+    ProcessedMail.objects.get_or_create(
         rule=rule,
         folder=rule.folder,
         uid=message_uid,
         uid_validity=uid_validity,
-        subject=message_subject,
-        received=make_aware(message_date) if is_naive(message_date) else message_date,
-        status="FAILED",
-        error=traceback.format_exc(),
+        defaults={
+            "subject": message_subject,
+            "received": received,
+            "status": "FAILED",
+            "error": traceback.format_exc(),
+        },
     )
 
 

src/paperless_mail/tests/test_mail.py

@@ -36,6 +36,7 @@ from paperless_mail.mail import MailAccountHandler
 from paperless_mail.mail import MailError
 from paperless_mail.mail import TagMailAction
 from paperless_mail.mail import apply_mail_action
+from paperless_mail.mail import error_callback
 from paperless_mail.mail import get_mailbox
 from paperless_mail.models import MailAccount
 from paperless_mail.models import MailRule
@@ -2045,6 +2046,44 @@ class TestPostConsumeAction(TestCase):
         self.assertIn("Test Exception", processed_mail.error)
 
 
+@pytest.mark.django_db
+class TestErrorCallback:
+    def test_error_callback_is_idempotent_for_same_mail(self) -> None:
+        """
+        GIVEN:
+            - A mail rule and a mail that failed to be consumed
+        WHEN:
+            - error_callback is invoked more than once for the same mail, as
+              happens when task_allow_error_cb_on_chord_header fires the
+              errback once per failed header task in a chord
+        THEN:
+            - Only one ProcessedMail row is created for that mail
+        """
+        rule = MailRuleFactory()
+        message_uid = "12345"
+
+        for _ in range(2):
+            error_callback(
+                None,
+                Exception("Test Exception"),
+                None,
+                rule_id=rule.pk,
+                message_uid=message_uid,
+                message_subject="Test Subject",
+                message_date=timezone.make_aware(
+                    timezone.datetime(2023, 1, 1, 12, 0, 0),
+                ),
+            )
+
+        processed_mails = ProcessedMail.objects.filter(
+            rule=rule,
+            uid=message_uid,
+            folder=rule.folder,
+        )
+        assert processed_mails.count() == 1
+        assert processed_mails.get().status == "FAILED"
+
+
 class TestManagementCommand(TestCase):
     @mock.patch(
         "paperless_mail.management.commands.mail_fetcher.tasks.process_mail_accounts",