Compare commits

..
Author SHA1 Message Date
stumpylogandClaude Sonnet 5 8de1d18762 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 <noreply@anthropic.com>
2026-09-08 17:00:04 -07:00
7 changed files with 142 additions and 402 deletions
-14
View File
@@ -374,7 +374,6 @@ class Document(SoftDeleteModel, ModelWithOwner): # type: ignore[django-manager-
If the queryset already annotated ``effective_content``, that value is used.
"""
# Here to avoid circular import
from documents.versioning import LATEST_VERSION_CONTENT_PREFETCH_ATTR
from documents.versioning import sort_versions_newest_first
from documents.versioning import versions_newest_first
@@ -384,19 +383,6 @@ class Document(SoftDeleteModel, ModelWithOwner): # type: ignore[django-manager-
if self.root_document_id is not None or self.pk is None:
return self.content
latest_version_prefetch = getattr(
self,
LATEST_VERSION_CONTENT_PREFETCH_ATTR,
None,
)
if latest_version_prefetch is not None:
# Empty list means prefetch ran and found no versions — use own content.
return (
latest_version_prefetch[0].content
if latest_version_prefetch
else self.content
)
prefetched_cache = getattr(self, "_prefetched_objects_cache", None)
prefetched_versions = (
prefetched_cache.get("versions")
+2 -9
View File
@@ -89,7 +89,6 @@ from documents.templating.utils import convert_format_str_to_template_format
from documents.templating.workflows import validate_workflow_template
from documents.validators import uri_validator
from documents.validators import url_validator
from documents.versioning import has_prefetched_effective_content
from documents.versioning import sort_versions_newest_first
if TYPE_CHECKING:
@@ -1153,14 +1152,8 @@ class DocumentSerializer(
def to_representation(self, instance):
doc = super().to_representation(instance)
if "content" in self.fields and has_prefetched_effective_content(instance):
# Only resolve version-aware content when it's cheap: an SQL
# annotation or a versions prefetch is already on the instance.
# A caller that set up neither (e.g. TrashView, GlobalSearchView,
# which build their own querysets) gets the document's own,
# unresolved content instead of paying for an extra per-instance
# query -- same as before effective_content resolution existed.
doc["content"] = instance.get_effective_content() or ""
if "content" in self.fields and hasattr(instance, "effective_content"):
doc["content"] = getattr(instance, "effective_content") or ""
if self.truncate_content and "content" in self.fields:
doc["content"] = doc.get("content")[0:550]
return doc
@@ -1,240 +0,0 @@
from __future__ import annotations
from types import SimpleNamespace
from typing import TYPE_CHECKING
import pytest
from django.db import connection
from django.test.utils import CaptureQueriesContext
from rest_framework import status
from documents.models import Document
from documents.tests.factories import DocumentFactory
from documents.versioning import LATEST_VERSION_CONTENT_PREFETCH_ATTR
from documents.versioning import has_prefetched_effective_content
from documents.versioning import latest_version_content_prefetch
from documents.views import DocumentViewSet
if TYPE_CHECKING:
from rest_framework.test import APIClient
class TestNeedsEffectiveContentAnnotation:
"""
DocumentViewSet._needs_effective_content_annotation() decides whether
the effective_content correlated subquery is worth attaching to the
queryset at all -- see TestDocumentListEffectiveContentAnnotation below
for why. This only checks that decision's own logic (a plain query-param
membership test), not that Django/DRF's filtering machinery works.
"""
@pytest.mark.parametrize(
("params", "expected"),
[
({}, False),
({"ordering": "-added"}, False),
({"tags__id__in": "1,2"}, False),
({"search": ""}, False),
({"search": " "}, False),
({"content__icontains": ""}, False),
({"search": "foo"}, True),
({"title_content": "foo"}, True),
({"content__istartswith": "foo"}, True),
({"content__iendswith": "foo"}, True),
({"content__icontains": "foo"}, True),
({"content__iexact": "foo"}, True),
],
)
def test_detects_content_filter_params(
self,
params: dict[str, str],
expected: bool, # noqa: FBT001
) -> None:
# GIVEN a view bound to a request carrying the given query params
view = DocumentViewSet()
view.request = SimpleNamespace(query_params=params)
# WHEN checking whether the effective_content annotation is needed
# THEN it's needed only for requests that actually filter on it
assert view._needs_effective_content_annotation() is expected
@pytest.mark.django_db
class TestDocumentListEffectiveContentAnnotation:
"""
DocumentViewSet.get_queryset() only attaches the effective_content
correlated subquery when a request actually filters on it. Attaching it
unconditionally re-executes it once per candidate row before the page's
LIMIT is applied -- fine on SQLite/Postgres, but pathological on
MariaDB's default cardinality estimation for the root_document_id
self-join once candidate counts get large (see the root_document_id /
effective_content perf investigation).
"""
def test_list_without_content_filter_skips_annotation_but_returns_latest_content(
self,
admin_client: APIClient,
) -> None:
# GIVEN a root document whose latest version has different content
root = DocumentFactory(content="old-root-content")
DocumentFactory(
root_document=root,
version_index=1,
content="new-version-content",
)
# WHEN listing documents with no search/content-filter param
with CaptureQueriesContext(connection) as ctx:
response = admin_client.get("/api/documents/?fields=id,content")
# THEN the response still reflects the latest version's content...
assert response.status_code == status.HTTP_200_OK
assert response.data["results"] == [
{"id": root.id, "content": "new-version-content"},
]
# ...without the database ever evaluating effective_content per row
assert not any(
"effective_content" in query["sql"] for query in ctx.captured_queries
)
def test_latest_version_content_prefetch_carries_only_the_newest_version(
self,
) -> None:
# GIVEN a root document with two versions
root = DocumentFactory(content="root-content")
DocumentFactory(
root_document=root,
version_index=1,
content="older-version-content",
)
DocumentFactory(
root_document=root,
version_index=2,
content="newest-version-content",
)
# WHEN fetching the root through latest_version_content_prefetch()
fetched_root = (
Document.objects.filter(pk=root.pk)
.prefetch_related(
latest_version_content_prefetch(),
)
.get()
)
# THEN the prefetch carries only the single newest version, not
# every historical version's content (the whole point of not
# reusing the metadata-only "versions" prefetch for this)
latest = getattr(fetched_root, LATEST_VERSION_CONTENT_PREFETCH_ATTR)
assert [v.content for v in latest] == ["newest-version-content"]
class TestHasPrefetchedEffectiveContent:
"""
DocumentSerializer.to_representation() only calls get_effective_content()
when has_prefetched_effective_content() says it's cheap -- otherwise a
caller that never set up an annotation or prefetch (TrashView,
GlobalSearchView, which build their own querysets and don't display
content at all) would pay for a per-instance query nobody asked for.
"""
def test_false_with_no_annotation_or_prefetch(self) -> None:
document = Document()
assert has_prefetched_effective_content(document) is False
def test_true_with_effective_content_annotation(self) -> None:
document = Document()
document.effective_content = "resolved"
assert has_prefetched_effective_content(document) is True
def test_true_with_lean_prefetch_attr_even_when_empty(self) -> None:
document = Document()
setattr(document, LATEST_VERSION_CONTENT_PREFETCH_ATTR, [])
assert has_prefetched_effective_content(document) is True
def test_true_with_metadata_versions_prefetch_cache(self) -> None:
document = Document()
document._prefetched_objects_cache = {"versions": []}
assert has_prefetched_effective_content(document) is True
def _get_effective_content_fallback_queries(
ctx: CaptureQueriesContext,
) -> list[dict[str, str]]:
"""
Document.get_effective_content()'s per-instance fallback (no annotation,
no prefetch) is a `.values_list("content", flat=True).first()` query --
a SELECT of just the content column. Distinct from get_versions()'s own,
unrelated per-instance metadata query (id/checksum/added/etc, no
content) run to build the "versions" response field, which isn't part
of what this test file covers.
"""
return [
q
for q in ctx.captured_queries
if q["sql"].startswith('SELECT "documents_document"."content" FROM')
]
@pytest.mark.django_db
class TestTrashAndGlobalSearchEffectiveContentIsNeverPerInstance:
"""
TrashView and GlobalSearchView serialize Document instances with
DocumentSerializer too, but build their querysets independently of
DocumentViewSet.get_queryset(). TrashView doesn't display content at all,
so it keeps the document's own unresolved content; GlobalSearchView
annotates effective_content itself, so it shows the latest version's.
Neither should ever fall back to a per-instance query.
"""
def test_trash_list_shows_unresolved_content_with_no_extra_query(
self,
admin_client: APIClient,
) -> None:
# GIVEN a trashed root document whose own content differs from what
# a (also trashed, since deletion cascades) version would have had
root = DocumentFactory(content="own-content")
DocumentFactory(
root_document=root,
version_index=1,
content="version-content",
)
root.delete()
# WHEN listing trash
with CaptureQueriesContext(connection) as ctx:
response = admin_client.get("/api/trash/")
# THEN the response shows the document's own content...
assert response.status_code == status.HTTP_200_OK
[result] = [r for r in response.data["results"] if r["id"] == root.id]
assert result["content"] == "own-content"
# ...without ever querying for versions to resolve it
assert _get_effective_content_fallback_queries(ctx) == []
def test_global_search_db_only_shows_latest_version_content_with_no_extra_query(
self,
admin_client: APIClient,
) -> None:
# GIVEN a root document, findable by title, whose own content
# differs from its latest version's
root = DocumentFactory(title="findme", content="own-content")
DocumentFactory(
root_document=root,
version_index=1,
content="version-content",
)
# WHEN using the global search endpoint's db_only mode
with CaptureQueriesContext(connection) as ctx:
response = admin_client.get(
"/api/search/?query=findme&db_only=true",
)
# THEN the response shows the latest version's content, resolved by
# GlobalSearchView's own effective_content annotation...
assert response.status_code == status.HTTP_200_OK
[result] = [d for d in response.data["documents"] if d["id"] == root.id]
assert result["content"] == "version-content"
# ...with no per-instance fallback query
assert _get_effective_content_fallback_queries(ctx) == []
-65
View File
@@ -7,12 +7,9 @@ from typing import Any
from django.db.models import F
from django.db.models import OuterRef
from django.db.models import Prefetch
from django.db.models import QuerySet
from django.db.models import Subquery
from django.db.models import Window
from django.db.models.functions import Coalesce
from django.db.models.functions import RowNumber
from documents.models import Document
@@ -49,68 +46,6 @@ def annotate_effective_content(documents: QuerySet[Document]) -> QuerySet[Docume
)
LATEST_VERSION_CONTENT_PREFETCH_ATTR = "_latest_version_content_prefetch"
def latest_version_content_prefetch() -> Prefetch:
"""
A Prefetch for Document.versions scoped to just the newest version's
content, for get_effective_content()'s fallback when no SQL annotation
is present.
Deliberately not merged into a metadata-only "versions" prefetch (the one
used for the serialized versions list): that one fetches every historical
version of every document, and pulling full OCR content for versions
nobody will read wastes DB transfer/memory at scale. This one is windowed
down to a single row per root, then bounded by Prefetch's own IN-list to
whatever page/result set it's attached to -- one cheap bulk query total,
not one per document and not one per version.
"""
return Prefetch(
"versions",
queryset=(
Document.objects.filter(
root_document_id__isnull=False,
deleted_at__isnull=True,
)
.annotate(
rn=Window(
RowNumber(),
partition_by=F("root_document_id"),
order_by=[
F("version_index").desc(nulls_last=True),
F("id").desc(),
],
),
)
.filter(rn=1)
.only("id", "root_document_id", "content")
),
to_attr=LATEST_VERSION_CONTENT_PREFETCH_ATTR,
)
def has_prefetched_effective_content(document: Document) -> bool:
"""
True if document.get_effective_content() can answer without an extra
per-instance query -- an SQL ``effective_content`` annotation, the lean
latest_version_content_prefetch(), or the metadata-only "versions"
prefetch is already present on the instance.
Callers that haven't set any of those up (e.g. views that build their
own querysets independently of DocumentViewSet.get_queryset(), like
TrashView or GlobalSearchView) intentionally don't pay for version-aware
content resolution -- see DocumentSerializer.to_representation(), which
uses this to decide whether to call get_effective_content() at all.
"""
if hasattr(document, "effective_content"):
return True
if getattr(document, LATEST_VERSION_CONTENT_PREFETCH_ATTR, None) is not None:
return True
prefetched_cache = getattr(document, "_prefetched_objects_cache", None)
return isinstance(prefetched_cache, dict) and "versions" in prefetched_cache
def sort_versions_newest_first(documents: list[Document]) -> list[Document]:
"""
Same sorting as versions_newest_first()
+28 -72
View File
@@ -36,6 +36,7 @@ from django.db.migrations.recorder import MigrationRecorder
from django.db.models import Avg
from django.db.models import Case
from django.db.models import Count
from django.db.models import F
from django.db.models import IntegerField
from django.db.models import Max
from django.db.models import Model
@@ -136,14 +137,12 @@ from documents.filters import CustomFieldFilterSet
from documents.filters import DocumentFilterSet
from documents.filters import DocumentsOrderingFilter
from documents.filters import DocumentTypeFilterSet
from documents.filters import EffectiveContentFilter
from documents.filters import PaperlessTaskFilterSet
from documents.filters import PermittedObjectsFilter
from documents.filters import ShareLinkBundleFilterSet
from documents.filters import ShareLinkFilterSet
from documents.filters import StoragePathFilterSet
from documents.filters import TagFilterSet
from documents.filters import TitleContentFilter
from documents.mail import EmailAttachment
from documents.mail import send_email
from documents.matching import match_correspondents
@@ -237,7 +236,6 @@ from documents.versioning import annotate_effective_content
from documents.versioning import get_latest_version_for_root
from documents.versioning import get_request_version_param
from documents.versioning import get_root_document
from documents.versioning import latest_version_content_prefetch
from documents.versioning import resolve_requested_version_for_root
from documents.versioning import versions_newest_first
from paperless import version
@@ -1085,49 +1083,12 @@ class DocumentViewSet(
],
}
@classmethod
def _content_filter_params(cls) -> tuple[str, ...]:
"""
Query params whose filtering needs effective_content evaluated in SQL
against every candidate row -- see
_needs_effective_content_annotation(). Derived rather than
hand-maintained so a new content-filtering param counts automatically.
"""
params = [
name
for name, f in DocumentFilterSet.declared_filters.items()
if isinstance(f, (TitleContentFilter, EffectiveContentFilter))
]
if "effective_content" in cls.search_fields:
params.append(SearchFilter().search_param)
return tuple(params)
def _needs_effective_content_annotation(self) -> bool:
# effective_content is a per-row correlated subquery resolving each
# document's latest version. Filtering *on* it forces the database to
# evaluate it for every candidate row before reaching the LIMIT, which
# the root_document_id self-join makes pathological on MariaDB
# specifically once real candidate counts get large; otherwise the
# "versions" prefetch + Document.get_effective_content() resolves only
# the page that survives pagination. Every param here is deprecated in
# favor of the Tantivy-backed search endpoint (see filters.py's
# TitleContentFilter/EffectiveContentFilter docs), so pay that cost
# only when one is actually used. Blank values don't count, matching
# how those filters themselves no-op on them -- an empty `?search=`
# applies no predicate.
params = self.request.query_params
return any(
params.get(param, "").strip() for param in self._content_filter_params()
)
def _needs_effective_content_prefetch(self) -> bool:
# The prefetch spares get_effective_content() a per-instance fallback
# query, but only earns itself when content can reach the response.
# Mirror get_serializer() below: no `fields` param keeps every field.
fields_param = self.request.query_params.get("fields", None)
return fields_param is None or "content" in fields_param.split(",")
def get_queryset(self):
latest_version_content = Subquery(
versions_newest_first(
Document.objects.filter(root_document=OuterRef("pk")),
).values("content")[:1],
)
# A correlated subquery avoids the LEFT JOIN + Count() this used to
# be, which forced a GROUP BY aggregate over every matching document
# before the query could even be sorted or limited.
@@ -1147,38 +1108,33 @@ class DocumentViewSet(
# ObjectFilter.filter(). A blanket .distinct() here forces the
# database to fully sort and dedupe every visible document before
# it can apply LIMIT, which is disastrous at scale.
prefetches = [
Prefetch(
"versions",
queryset=Document.objects.only(
"id",
"added",
"checksum",
"version_label",
"root_document_id",
"version_index",
),
),
"tags",
Prefetch(
"custom_fields",
queryset=CustomFieldInstance.objects.select_related("field"),
),
# NotesSerializer nests the author, this avoids query per note
Prefetch("notes", queryset=Note.objects.select_related("user")),
]
if self._needs_effective_content_prefetch():
prefetches.append(latest_version_content_prefetch())
queryset = (
return (
Document.objects.filter(root_document__isnull=True)
.order_by("-created", "-id")
.annotate(effective_content=Coalesce(latest_version_content, F("content")))
.annotate(num_notes=Coalesce(note_count, 0))
.select_related("correspondent", "storage_path", "document_type", "owner")
.prefetch_related(*prefetches)
.prefetch_related(
Prefetch(
"versions",
queryset=Document.objects.only(
"id",
"added",
"checksum",
"version_label",
"root_document_id",
"version_index",
),
),
"tags",
Prefetch(
"custom_fields",
queryset=CustomFieldInstance.objects.select_related("field"),
),
# NotesSerializer nests the author, this avoids query per note
Prefetch("notes", queryset=Note.objects.select_related("user")),
)
)
if self._needs_effective_content_annotation():
queryset = annotate_effective_content(queryset)
return queryset
def get_serializer(self, *args, **kwargs):
fields_param = self.request.query_params.get("fields", None)
+23 -2
View File
@@ -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)
@@ -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."