diff --git a/src/documents/consumer.py b/src/documents/consumer.py index bb60d21085..5110fbd843 100644 --- a/src/documents/consumer.py +++ b/src/documents/consumer.py @@ -15,7 +15,6 @@ from django.db import transaction from django.db.models import Max from django.db.models import Q from django.utils import timezone -from filelock import FileLock from rest_framework.reverse import reverse from documents.barcodes import read_barcode_values @@ -28,6 +27,7 @@ from documents.file_handling import generate_filename from documents.file_handling import generate_unique_filename from documents.file_handling import validate_path_in_root from documents.loggers import LoggingMixin +from documents.media_lock import media_lock from documents.models import Correspondent from documents.models import CustomField from documents.models import CustomFieldInstance @@ -695,7 +695,7 @@ class ConsumerPlugin( # After everything is in the database, copy the files into # place. If this fails, we'll also rollback the transaction. - with FileLock(settings.MEDIA_LOCK): + with media_lock(): generated_filename = generate_unique_filename(document) if ( len(str(generated_filename)) diff --git a/src/documents/file_moves.py b/src/documents/file_moves.py new file mode 100644 index 0000000000..7ca349174f --- /dev/null +++ b/src/documents/file_moves.py @@ -0,0 +1,31 @@ +""" +Document file moves, coalesced to once per unit of work. + +Every save that can change a document's generated filename asks for a file +move through request_file_move(). Inside a scope opened by defer_file_moves() +(a request, a Celery task, the consumer, or a per-item command step) the +request is only recorded, and the outermost scope moves each recorded document +once, after its own writes have committed. Outside any scope the move happens +immediately. + +The mover always takes media.lock first and then reads the document fresh, so +it starts from the latest committed state and never waits on media.lock while +its own transaction holds document rows. +""" + +from __future__ import annotations + +from django.db import transaction + + +def in_user_transaction() -> bool: + """ + True when the default connection is inside an atomic block that was not + created by Django's test case machinery. + + Django marks the atomic blocks TestCase (and pytest-django) wrap around + each test with _from_testcase, the same flag it uses for durable checks. + Ignoring them keeps tests on the same path as production. + """ + connection = transaction.get_connection() + return any(not block._from_testcase for block in connection.atomic_blocks) diff --git a/src/documents/mail.py b/src/documents/mail.py index e9c4cf6adc..7b540d480e 100644 --- a/src/documents/mail.py +++ b/src/documents/mail.py @@ -4,9 +4,9 @@ from dataclasses import dataclass from email import message_from_bytes from pathlib import Path -from django.conf import settings from django.core.mail import EmailMessage -from filelock import FileLock + +from documents.media_lock import media_lock @dataclass(frozen=True) @@ -48,7 +48,7 @@ def send_email( used_filenames: set[str] = set() # Something could be renaming the file concurrently so it can't be attached - with FileLock(settings.MEDIA_LOCK): + with media_lock(): for attachment in attachments: filename = _get_unique_filename( attachment.friendly_name, diff --git a/src/documents/management/commands/document_exporter.py b/src/documents/management/commands/document_exporter.py index 10b929679d..6f03fe72e2 100644 --- a/src/documents/management/commands/document_exporter.py +++ b/src/documents/management/commands/document_exporter.py @@ -18,10 +18,11 @@ from django.core import serializers from django.core.management.base import CommandError from django.db import transaction from django.utils import timezone -from filelock import FileLock from guardian.models import GroupObjectPermission from guardian.models import UserObjectPermission +from documents.media_lock import media_lock + if TYPE_CHECKING: from collections.abc import Generator @@ -328,7 +329,7 @@ class Command(CryptMixin, PaperlessCommand): ) # Prevent any ongoing changes in the documents while exporting - with FileLock(settings.MEDIA_LOCK), sink: + with media_lock(), sink: self.dump(sink) def dump(self, sink: ExportSink) -> None: diff --git a/src/documents/management/commands/document_importer.py b/src/documents/management/commands/document_importer.py index fa0e4bd330..d94bb72870 100644 --- a/src/documents/management/commands/document_importer.py +++ b/src/documents/management/commands/document_importer.py @@ -29,7 +29,6 @@ from django.db.models import GeneratedField from django.db.models import Model from django.db.models.signals import m2m_changed from django.db.models.signals import post_save -from filelock import FileLock from guardian.shortcuts import clear_ct_cache from documents.export.compression import compress_type_readable @@ -37,6 +36,7 @@ from documents.export.compression import unreadable_method_names from documents.file_handling import create_source_path_directory from documents.management.commands.base import PaperlessCommand from documents.management.commands.mixins import CryptMixin +from documents.media_lock import media_lock from documents.models import Correspondent from documents.models import CustomField from documents.models import CustomFieldInstance @@ -644,7 +644,7 @@ class Command(CryptMixin, PaperlessCommand): else: archive_path = None - with FileLock(settings.MEDIA_LOCK): + with media_lock(): if Path(document.source_path).is_file(): raise FileExistsError(document.source_path) @@ -678,7 +678,7 @@ class Command(CryptMixin, PaperlessCommand): f"Share link bundle {bundle.pk} does not have a valid file path.", ) - with FileLock(settings.MEDIA_LOCK): + with media_lock(): bundle_target_path.parent.mkdir(parents=True, exist_ok=True) copy_file_with_basic_stats( bundle_source_path, diff --git a/src/documents/media_lock.py b/src/documents/media_lock.py new file mode 100644 index 0000000000..8674e40e7d --- /dev/null +++ b/src/documents/media_lock.py @@ -0,0 +1,36 @@ +""" +The global media lock, reentrant within a thread. + +Code that both needs media.lock and opens a transaction that locks document +rows must take media.lock first, so no transaction ever waits on it while +holding rows. Reentrancy lets code already holding the lock (for example the +consumer) call into code that takes it again (for example a workflow email +action attaching files) without blocking itself. +""" + +from __future__ import annotations + +from contextlib import contextmanager +from typing import TYPE_CHECKING + +from django.conf import settings +from filelock import FileLock + +if TYPE_CHECKING: + from collections.abc import Iterator + + +@contextmanager +def media_lock(timeout: float = -1) -> Iterator[None]: + """ + Hold settings.MEDIA_LOCK for the block. + + is_singleton returns one FileLock instance per path in this process, and + the default thread_local=True keeps the hold count per thread, so the + same thread re-enters while other threads and processes are excluded. + The timeout goes to acquire() because a singleton rejects construction + with different parameters. A negative timeout waits forever. + """ + lock = FileLock(settings.MEDIA_LOCK, is_singleton=True) + with lock.acquire(timeout=timeout): + yield diff --git a/src/documents/signals/handlers.py b/src/documents/signals/handlers.py index 9c20e97f21..76dc65f9cd 100644 --- a/src/documents/signals/handlers.py +++ b/src/documents/signals/handlers.py @@ -26,7 +26,6 @@ from django.db import models from django.db.models import Q from django.dispatch import receiver from django.utils import timezone -from filelock import FileLock from rest_framework import serializers from documents import matching @@ -38,6 +37,7 @@ from documents.file_handling import create_source_path_directory from documents.file_handling import delete_empty_directories from documents.file_handling import generate_filename from documents.file_handling import generate_unique_filename +from documents.media_lock import media_lock from documents.models import Correspondent from documents.models import CustomField from documents.models import CustomFieldInstance @@ -339,7 +339,7 @@ def set_storage_path( # see empty_trash in documents/tasks.py for signal handling def cleanup_document_deletion(sender, instance, **kwargs) -> None: - with FileLock(settings.MEDIA_LOCK): + with media_lock(): if settings.EMPTY_TRASH_DIR: # Find a non-conflicting filename in case a document with the same # name was moved to trash earlier @@ -475,7 +475,7 @@ def update_filename_and_move_files( # This will in turn cause this logic to move the file where it belongs. return - with FileLock(settings.MEDIA_LOCK): + with media_lock(): # If this was waiting for the lock, the filename or archive_filename # of this document may have been updated. This happens if multiple updates # get queued from the UI for the same document diff --git a/src/documents/tasks.py b/src/documents/tasks.py index 8f0e85f165..ac6db4efb2 100644 --- a/src/documents/tasks.py +++ b/src/documents/tasks.py @@ -16,7 +16,6 @@ from django.db import models from django.db import transaction from django.db.models.signals import post_save from django.utils import timezone -from filelock import FileLock from documents import sanity_checker from documents.barcodes import BarcodePlugin @@ -42,6 +41,7 @@ from documents.double_sided import CollatePlugin from documents.file_handling import create_source_path_directory from documents.file_handling import generate_unique_filename from documents.matching import prefilter_documents_by_workflowtrigger +from documents.media_lock import media_lock from documents.models import Correspondent from documents.models import CustomFieldInstance from documents.models import Document @@ -482,7 +482,7 @@ def update_document_content_maybe_archive_file( modified=timezone.now(), ) - with FileLock(settings.MEDIA_LOCK): + with media_lock(): if parser.get_archive_path(): create_source_path_directory(document.archive_path) shutil.move(parser.get_archive_path(), document.archive_path) diff --git a/src/documents/tests/test_media_lock.py b/src/documents/tests/test_media_lock.py new file mode 100644 index 0000000000..d76dbdfef4 --- /dev/null +++ b/src/documents/tests/test_media_lock.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +import threading +from typing import TYPE_CHECKING + +import pytest +from django.db import transaction +from filelock import Timeout + +from documents.file_moves import in_user_transaction +from documents.media_lock import media_lock +from paperless_testing.locks import record_media_lock_acquisitions + +if TYPE_CHECKING: + from paperless_testing.dirs import PaperlessDirs + + +@pytest.mark.usefixtures("paperless_dirs") +class TestMediaLock: + def test_same_thread_reenters_without_blocking(self) -> None: + """ + GIVEN: + - media.lock held by the current thread + WHEN: + - The same thread acquires it again with a zero timeout + THEN: + - The nested acquisition succeeds immediately + """ + with media_lock(), media_lock(timeout=0): + pass + + def test_other_thread_is_excluded(self) -> None: + """ + GIVEN: + - media.lock held by the current thread + WHEN: + - Another thread tries to acquire it with a short timeout + THEN: + - The other thread times out + - After release, the other thread can acquire it + """ + outcomes: list[str] = [] + + def try_acquire() -> None: + try: + with media_lock(timeout=0.2): + outcomes.append("acquired") + except Timeout: + outcomes.append("timeout") + + with media_lock(): + worker = threading.Thread(target=try_acquire) + worker.start() + worker.join() + + worker = threading.Thread(target=try_acquire) + worker.start() + worker.join() + + assert outcomes == ["timeout", "acquired"] + + def test_lock_file_is_the_configured_path( + self, + paperless_dirs: PaperlessDirs, + ) -> None: + """ + GIVEN: + - The test's media directory layout + WHEN: + - media.lock is acquired + THEN: + - The lock file is created at settings.MEDIA_LOCK + """ + with media_lock(): + assert paperless_dirs.media_lock.exists() + + +@pytest.mark.django_db +class TestInUserTransaction: + def test_false_inside_test_case_wrapper(self) -> None: + """ + GIVEN: + - The transaction pytest-django wraps around the test + WHEN: + - in_user_transaction is called + THEN: + - It returns False + """ + assert in_user_transaction() is False + + def test_true_inside_atomic(self) -> None: + """ + GIVEN: + - An atomic block opened by code under test + WHEN: + - in_user_transaction is called inside it + THEN: + - It returns True + """ + with transaction.atomic(): + assert in_user_transaction() is True + + +@pytest.mark.django_db +@pytest.mark.usefixtures("paperless_dirs") +class TestRecordMediaLockAcquisitions: + def test_records_only_first_acquisitions(self) -> None: + """ + GIVEN: + - Nested media.lock acquisitions, the outer one inside a transaction + WHEN: + - They are recorded + THEN: + - Only the outer, first acquisition is recorded, flagged as inside + a transaction + """ + with record_media_lock_acquisitions() as acquisitions: + with transaction.atomic(), media_lock(), media_lock(): + pass + + assert [a.in_user_transaction for a in acquisitions] == [True] diff --git a/src/paperless_testing/locks.py b/src/paperless_testing/locks.py new file mode 100644 index 0000000000..b610d15113 --- /dev/null +++ b/src/paperless_testing/locks.py @@ -0,0 +1,48 @@ +"""Record first acquisitions of the global media lock during a test.""" + +from __future__ import annotations + +from contextlib import contextmanager +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING +from unittest import mock + +if TYPE_CHECKING: + from collections.abc import Iterator + + +@dataclass(frozen=True, slots=True) +class MediaLockAcquisition: + """One first (non-reentrant) acquisition of settings.MEDIA_LOCK.""" + + in_user_transaction: bool + + +@contextmanager +def record_media_lock_acquisitions() -> Iterator[list[MediaLockAcquisition]]: + """ + Record every first acquisition of settings.MEDIA_LOCK made in the block. + + Reentrant acquisitions by a thread that already holds the lock never wait, + so they are not recorded. Each record notes whether a real (non test case) + transaction was open at that moment, which is the lock ordering hazard. + Usable from pytest tests and unittest TestCase classes alike. + """ + from django.conf import settings + from filelock import BaseFileLock + + from documents.file_moves import in_user_transaction + + records: list[MediaLockAcquisition] = [] + original_acquire = BaseFileLock.acquire + + def recording_acquire(self: BaseFileLock, *args: object, **kwargs: object): + if Path(self.lock_file) == Path(settings.MEDIA_LOCK) and not self.is_locked: + records.append( + MediaLockAcquisition(in_user_transaction=in_user_transaction()), + ) + return original_acquire(self, *args, **kwargs) + + with mock.patch.object(BaseFileLock, "acquire", recording_acquire): + yield records