From 8de1d187629674fe3410c853bf316c07a3c9907b Mon Sep 17 00:00:00 2001 From: stumpylog <797416+stumpylog@users.noreply.github.com> Date: Tue, 8 Sep 2026 17:00:04 -0700 Subject: [PATCH] Fix: prevent overlapping mail-account processing runs process_mail_accounts had no guard against a scheduled run still being in progress when the next one fires (e.g. a large attachment batch taking longer than the check interval). ProcessedMail dedup only records a message once handling finishes, so an overlapping run could still pick up the same not-yet-recorded message. Skip a run outright if another MAIL_FETCH task is already PENDING/STARTED. Co-Authored-By: Claude Sonnet 5 --- src/paperless_mail/tasks.py | 25 +++++- .../test_process_mail_accounts_overlap.py | 89 +++++++++++++++++++ 2 files changed, 112 insertions(+), 2 deletions(-) create mode 100644 src/paperless_mail/tests/test_process_mail_accounts_overlap.py diff --git a/src/paperless_mail/tasks.py b/src/paperless_mail/tasks.py index df1f30d91..29103be37 100644 --- a/src/paperless_mail/tasks.py +++ b/src/paperless_mail/tasks.py @@ -1,7 +1,9 @@ import logging +from celery import Task from celery import shared_task +from documents.models import PaperlessTask from paperless_mail.mail import MailAccountHandler from paperless_mail.mail import MailError from paperless_mail.models import MailAccount @@ -10,8 +12,27 @@ from paperless_mail.models import MailRule logger = logging.getLogger("paperless.mail.tasks") -@shared_task -def process_mail_accounts(account_ids: list[int] | None = None) -> str: +@shared_task(bind=True) +def process_mail_accounts(self: Task, account_ids: list[int] | None = None) -> str: + # A scheduled check can still be running (or queued) when the next one + # fires, e.g. a large attachment batch that takes longer to process than + # the check interval. ProcessedMail dedup only records a message once its + # handling has finished, so an overlapping run can still pick up the same + # not-yet-recorded message. Skip outright rather than race it. + other_mail_fetch_running = ( + PaperlessTask.objects.filter( + task_type=PaperlessTask.TaskType.MAIL_FETCH, + status__in=[PaperlessTask.Status.PENDING, PaperlessTask.Status.STARTED], + ) + .exclude(task_id=self.request.id) + .exists() + ) + if other_mail_fetch_running: + logger.info( + "Mail account processing is already running; skipping this run.", + ) + return "Skipped: mail account processing already in progress." + total_new_documents = 0 accounts = ( MailAccount.objects.filter(pk__in=account_ids) diff --git a/src/paperless_mail/tests/test_process_mail_accounts_overlap.py b/src/paperless_mail/tests/test_process_mail_accounts_overlap.py new file mode 100644 index 000000000..1acf7a2ae --- /dev/null +++ b/src/paperless_mail/tests/test_process_mail_accounts_overlap.py @@ -0,0 +1,89 @@ +from unittest import mock + +import pytest + +from documents.models import PaperlessTask +from paperless_mail import tasks +from paperless_mail.tests.factories import MailAccountFactory +from paperless_mail.tests.factories import MailRuleFactory + + +@pytest.mark.django_db +class TestProcessMailAccountsOverlap: + def test_skips_when_another_mail_fetch_task_is_running(self) -> None: + account = MailAccountFactory.create() + MailRuleFactory.create(account=account, enabled=True) + + PaperlessTask.objects.create( + task_id="other-running-task", + task_type=PaperlessTask.TaskType.MAIL_FETCH, + trigger_source=PaperlessTask.TriggerSource.SCHEDULED, + status=PaperlessTask.Status.STARTED, + ) + + with mock.patch.object( + tasks.MailAccountHandler, + "handle_mail_account", + ) as mocked_handle: + result = tasks.process_mail_accounts() + + mocked_handle.assert_not_called() + assert result == "Skipped: mail account processing already in progress." + + def test_runs_when_no_other_mail_fetch_task_is_running(self) -> None: + account = MailAccountFactory.create() + MailRuleFactory.create(account=account, enabled=True) + + with mock.patch.object( + tasks.MailAccountHandler, + "handle_mail_account", + return_value=0, + ) as mocked_handle: + result = tasks.process_mail_accounts() + + mocked_handle.assert_called_once() + assert result == "No new documents were added." + + def test_ignores_completed_mail_fetch_tasks(self) -> None: + account = MailAccountFactory.create() + MailRuleFactory.create(account=account, enabled=True) + + PaperlessTask.objects.create( + task_id="finished-task", + task_type=PaperlessTask.TaskType.MAIL_FETCH, + trigger_source=PaperlessTask.TriggerSource.SCHEDULED, + status=PaperlessTask.Status.SUCCESS, + ) + + with mock.patch.object( + tasks.MailAccountHandler, + "handle_mail_account", + return_value=0, + ) as mocked_handle: + result = tasks.process_mail_accounts() + + mocked_handle.assert_called_once() + assert result == "No new documents were added." + + def test_does_not_skip_due_to_its_own_task_row(self) -> None: + account = MailAccountFactory.create() + MailRuleFactory.create(account=account, enabled=True) + + PaperlessTask.objects.create( + task_id="self-task-id", + task_type=PaperlessTask.TaskType.MAIL_FETCH, + trigger_source=PaperlessTask.TriggerSource.SCHEDULED, + status=PaperlessTask.Status.STARTED, + ) + + with mock.patch.object( + tasks.MailAccountHandler, + "handle_mail_account", + return_value=0, + ) as mocked_handle: + result = tasks.process_mail_accounts.apply( + task_id="self-task-id", + ).result + + mocked_handle.assert_called_once() + assert result == "No new documents were added."