Fix: scope ProcessedMail UID deduplication by IMAP UIDVALIDITY (#12839)

src/paperless_mail/mail.py

@@ -18,6 +18,7 @@ from celery import shared_task
 from celery.canvas import Signature
 from django.conf import settings
 from django.db import DatabaseError
+from django.db.models import Q
 from django.utils import timezone
 from django.utils.timezone import is_naive
 from django.utils.timezone import make_aware
@@ -245,6 +246,7 @@ def apply_mail_action(
     message_uid: str,
     message_subject: str,
     message_date: datetime.datetime,
+    uid_validity: str | None = None,
 ) -> None:
     """
     This shared task applies the mail action of a particular mail rule to the
@@ -286,6 +288,7 @@ def apply_mail_action(
             rule=rule,
             folder=rule.folder,
             uid=message_uid,
+            uid_validity=uid_validity,
             subject=message_subject,
             received=message_date,
             status="SUCCESS",
@@ -297,6 +300,7 @@ def apply_mail_action(
             rule=rule,
             folder=rule.folder,
             uid=message_uid,
+            uid_validity=uid_validity,
             subject=message_subject,
             received=message_date,
             status="FAILED",
@@ -314,6 +318,7 @@ def error_callback(
     message_uid: str,
     message_subject: str,
     message_date: datetime.datetime,
+    uid_validity: str | None = None,
 ) -> None:
     """
     A shared task that is called whenever something goes wrong during
@@ -325,6 +330,7 @@ def error_callback(
         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",
@@ -337,6 +343,7 @@ def queue_consumption_tasks(
     consume_tasks: list[Signature],
     rule: MailRule,
     message: MailMessage,
+    uid_validity: str | None,
 ) -> None:
     """
     Queue a list of consumption tasks (Signatures for the consume_file shared
@@ -348,6 +355,7 @@ def queue_consumption_tasks(
         message_uid=message.uid,
         message_subject=message.subject,
         message_date=message.date,
+        uid_validity=uid_validity,
     )
     chord(header=consume_tasks, body=mail_action_task).on_error(
         error_callback.s(
@@ -355,6 +363,7 @@ def queue_consumption_tasks(
             message_uid=message.uid,
             message_subject=message.subject,
             message_date=message.date,
+            uid_validity=uid_validity,
         ),
     ).delay()
 
@@ -460,6 +469,7 @@ class MailAccountHandler(LoggingMixin):
         super().__init__()
         self.renew_logging_group()
         self._init_preprocessors()
+        self._current_uid_validity: str | None = None
 
     def _init_preprocessors(self) -> None:
         self._message_preprocessors: list[MailMessagePreprocessor] = []
@@ -510,6 +520,21 @@ class MailAccountHandler(LoggingMixin):
                 "Unknown title selector.",
             )  # pragma: no cover
 
+    def _get_uid_validity(self, M: MailBox, folder: str) -> str | None:
+        try:
+            uid_validity = M.folder.status(folder, ["UIDVALIDITY"]).get("UIDVALIDITY")
+            if uid_validity is not None:
+                return str(uid_validity)
+        except errors.MailboxFolderStatusError as e:
+            self.log.warning(
+                f"Server does not support retrieving UIDVALIDITY for folder {folder}: {e}",
+            )
+        except Exception as e:
+            self.log.warning(
+                f"Unable to retrieve UIDVALIDITY for folder {folder}: {e}",
+            )
+        return None
+
     def _get_correspondent(
         self,
         message: MailMessage,
@@ -647,6 +672,8 @@ class MailAccountHandler(LoggingMixin):
                 f"does not exist in account {rule.account}",
             ) from err
 
+        self._current_uid_validity = self._get_uid_validity(M, rule.folder)
+
         criterias = make_criterias(rule, supports_gmail_labels=supports_gmail_labels)
 
         self.log.debug(
@@ -687,11 +714,17 @@ class MailAccountHandler(LoggingMixin):
                 )
                 continue
 
-            if ProcessedMail.objects.filter(
+            already_processed = ProcessedMail.objects.filter(
                 rule=rule,
                 uid=message.uid,
                 folder=rule.folder,
-            ).exists():
+            )
+            if self._current_uid_validity is not None:
+                already_processed = already_processed.filter(
+                    Q(uid_validity=self._current_uid_validity)
+                    | Q(uid_validity__isnull=True),
+                )
+            if already_processed.exists():
                 self.log.debug(
                     f"Skipping mail '{message.uid}' subject '{message.subject}' from '{message.from_}', already processed.",
                 )
@@ -921,6 +954,7 @@ class MailAccountHandler(LoggingMixin):
                 consume_tasks=consume_tasks,
                 rule=rule,
                 message=message,
+                uid_validity=self._current_uid_validity,
             )
         else:
             # No files to consume, just mark as processed if it wasn't by .eml processing
@@ -928,11 +962,13 @@ class MailAccountHandler(LoggingMixin):
                 rule=rule,
                 uid=message.uid,
                 folder=rule.folder,
+                uid_validity=self._current_uid_validity,
             ).exists():
                 ProcessedMail.objects.create(
                     rule=rule,
                     folder=rule.folder,
                     uid=message.uid,
+                    uid_validity=self._current_uid_validity,
                     subject=message.subject,
                     received=make_aware(message.date)
                     if is_naive(message.date)
@@ -1009,6 +1045,7 @@ class MailAccountHandler(LoggingMixin):
             consume_tasks=[consume_task],
             rule=rule,
             message=message,
+            uid_validity=self._current_uid_validity,
         )
 
         processed_elements = 1

src/paperless_mail/migrations/0004_processedmail_uid_validity.py

@@ -0,0 +1,22 @@
+from django.db import migrations
+from django.db import models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("paperless_mail", "0003_mailrule_stop_processing"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="processedmail",
+            name="uid_validity",
+            field=models.CharField(
+                blank=True,
+                editable=False,
+                max_length=64,
+                null=True,
+                verbose_name="uid validity",
+            ),
+        ),
+    ]

src/paperless_mail/models.py

@@ -338,6 +338,14 @@ class ProcessedMail(document_models.ModelWithOwner):
         editable=False,
     )
 
+    uid_validity = models.CharField(
+        _("uid validity"),
+        null=True,
+        blank=True,
+        max_length=64,
+        editable=False,
+    )
+
     subject = models.CharField(
         _("subject"),
         null=False,

src/paperless_mail/tests/test_mail.py

@@ -54,12 +54,16 @@ class _AttachmentDef:
 
 class BogusFolderManager:
     current_folder = "INBOX"
+    uidvalidity = "1"
 
     def set(self, new_folder) -> None:
         if new_folder not in ["INBOX", "spam"]:
             raise MailboxFolderSelectError(None, "uhm")
         self.current_folder = new_folder
 
+    def status(self, folder, options):
+        return {"UIDVALIDITY": self.uidvalidity}
+
 
 class BogusClient:
     def __init__(self, messages) -> None:
@@ -944,6 +948,223 @@ class TestMail(
 
         self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 1)
 
+    def test_handle_mail_account_skips_mail_already_processed_in_same_uidvalidity(
+        self,
+    ) -> None:
+        """
+        GIVEN:
+            - A ProcessedMail row recorded under the mailbox's current UIDVALIDITY
+        WHEN:
+            - A mail with the same UID is fetched from the same UIDVALIDITY epoch
+        THEN:
+            - The mail is skipped as a duplicate.
+        """
+        account = MailAccount.objects.create(
+            name="test",
+            imap_server="",
+            username="admin",
+            password="secret",
+        )
+        rule = MailRule.objects.create(
+            name="testrule",
+            account=account,
+            action=MailRule.MailAction.DELETE,
+        )
+
+        message = self.mailMocker.messageBuilder.create_message()
+        self.mailMocker.bogus_mailbox.messages = [message]
+        self.mailMocker.bogus_mailbox.updateClient()
+
+        self.mailMocker.bogus_mailbox.folder.uidvalidity = "SAME"
+        ProcessedMail.objects.create(
+            rule=rule,
+            folder=rule.folder,
+            uid=message.uid,
+            uid_validity="SAME",
+            subject="Previously processed mail",
+            status="SUCCESS",
+            received=timezone.make_aware(timezone.datetime(2023, 1, 1, 12, 0, 0)),
+        )
+
+        self.mail_account_handler.handle_mail_account(account)
+
+        self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 0)
+
+    def test_handle_mail_account_processes_mail_after_uidvalidity_change(
+        self,
+    ) -> None:
+        """
+        GIVEN:
+            - A ProcessedMail row recorded under a previous UIDVALIDITY epoch
+        WHEN:
+            - A mail with the same UID is fetched after UIDVALIDITY has changed
+        THEN:
+            - The mail is processed, not skipped as a duplicate.
+        """
+        account = MailAccount.objects.create(
+            name="test",
+            imap_server="",
+            username="admin",
+            password="secret",
+        )
+        rule = MailRule.objects.create(
+            name="testrule",
+            account=account,
+            action=MailRule.MailAction.DELETE,
+        )
+
+        message = self.mailMocker.messageBuilder.create_message()
+        self.mailMocker.bogus_mailbox.messages = [message]
+        self.mailMocker.bogus_mailbox.updateClient()
+
+        self.mailMocker.bogus_mailbox.folder.uidvalidity = "NEW"
+        ProcessedMail.objects.create(
+            rule=rule,
+            folder=rule.folder,
+            uid=message.uid,
+            uid_validity="OLD",
+            subject="Previously processed mail",
+            status="SUCCESS",
+            received=timezone.make_aware(timezone.datetime(2023, 1, 1, 12, 0, 0)),
+        )
+
+        self.mail_account_handler.handle_mail_account(account)
+
+        self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 1)
+
+    def test_handle_mail_account_skips_mail_processed_before_uidvalidity_tracking(
+        self,
+    ) -> None:
+        """
+        GIVEN:
+            - A ProcessedMail row recorded before UIDVALIDITY tracking existed
+              (uid_validity is NULL)
+        WHEN:
+            - A mail with the same UID is fetched
+        THEN:
+            - The mail is skipped as a duplicate, to avoid re-ingesting all
+              previously processed mail after upgrading.
+        """
+        account = MailAccount.objects.create(
+            name="test",
+            imap_server="",
+            username="admin",
+            password="secret",
+        )
+        rule = MailRule.objects.create(
+            name="testrule",
+            account=account,
+            action=MailRule.MailAction.DELETE,
+        )
+
+        message = self.mailMocker.messageBuilder.create_message()
+        self.mailMocker.bogus_mailbox.messages = [message]
+        self.mailMocker.bogus_mailbox.updateClient()
+
+        ProcessedMail.objects.create(
+            rule=rule,
+            folder=rule.folder,
+            uid=message.uid,
+            uid_validity=None,
+            subject="Previously processed mail",
+            status="SUCCESS",
+            received=timezone.make_aware(timezone.datetime(2023, 1, 1, 12, 0, 0)),
+        )
+
+        self.mail_account_handler.handle_mail_account(account)
+
+        self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 0)
+
+    def test_handle_mail_account_processes_mail_when_uidvalidity_unavailable(
+        self,
+    ) -> None:
+        """
+        GIVEN:
+            - The mail server fails to report a UIDVALIDITY for the folder
+        WHEN:
+            - A mail account is processed
+        THEN:
+            - The failure is logged and the rule still processes the mail,
+              instead of the whole rule being disabled.
+        """
+        account = MailAccount.objects.create(
+            name="test",
+            imap_server="",
+            username="admin",
+            password="secret",
+        )
+        _ = MailRule.objects.create(
+            name="testrule",
+            account=account,
+            action=MailRule.MailAction.DELETE,
+        )
+
+        message = self.mailMocker.messageBuilder.create_message()
+        self.mailMocker.bogus_mailbox.messages = [message]
+        self.mailMocker.bogus_mailbox.updateClient()
+
+        self.mailMocker.bogus_mailbox.folder.status = mock.MagicMock(
+            side_effect=errors.MailboxFolderStatusError(("NO", [b"unsupported"]), "OK"),
+        )
+
+        with self.assertLogs("paperless_mail", level="WARNING") as cm:
+            self.mail_account_handler.handle_mail_account(account)
+
+        self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 1)
+        self.assertEqual(len(cm.output), 1)
+        self.assertIn(
+            "Server does not support retrieving UIDVALIDITY",
+            cm.output[0],
+        )
+
+    def test_handle_mail_account_skips_mail_when_uidvalidity_unavailable_but_prior_record_exists(
+        self,
+    ) -> None:
+        """
+        GIVEN:
+            - A ProcessedMail row recorded with a real uid_validity value
+            - The mail server fails to report UIDVALIDITY (MailboxFolderStatusError),
+              so _get_uid_validity returns None
+        WHEN:
+            - A mail with the same UID is fetched
+        THEN:
+            - The mail is skipped as already-processed rather than re-ingested,
+              falling back to (rule, uid, folder) matching.
+        """
+        account = MailAccount.objects.create(
+            name="test",
+            imap_server="",
+            username="admin",
+            password="secret",
+        )
+        rule = MailRule.objects.create(
+            name="testrule",
+            account=account,
+            action=MailRule.MailAction.DELETE,
+        )
+
+        message = self.mailMocker.messageBuilder.create_message()
+        self.mailMocker.bogus_mailbox.messages = [message]
+        self.mailMocker.bogus_mailbox.updateClient()
+
+        ProcessedMail.objects.create(
+            rule=rule,
+            folder=rule.folder,
+            uid=message.uid,
+            uid_validity="REAL_VALIDITY",
+            subject="Previously processed mail",
+            status="SUCCESS",
+            received=timezone.make_aware(timezone.datetime(2023, 1, 1, 12, 0, 0)),
+        )
+
+        self.mailMocker.bogus_mailbox.folder.status = mock.MagicMock(
+            side_effect=errors.MailboxFolderStatusError(("NO", [b"unsupported"]), "OK"),
+        )
+
+        self.mail_account_handler.handle_mail_account(account)
+
+        self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 0)
+
     @pytest.mark.flaky(reruns=4)
     def test_handle_mail_account_flag(self) -> None:
         account = MailAccount.objects.create(