From 61b9bc336771def9b9383d3b6bcf74ea521fe01f Mon Sep 17 00:00:00 2001 From: stumpylog <797416+stumpylog@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:31:01 -0800 Subject: [PATCH] Fix: take media.lock before the consumer's storage transaction Consuming a new version locked the root document row with select_for_update and then waited on media.lock to place the file, the same lock ordering that let concurrent saves deadlock. On MariaDB even the version row's foreign key check holds a shared lock on the root. The consumer's storage phase now takes media.lock before opening its transaction and runs inside its own file move scope. File placement and workflow email actions inside the transaction re-enter the lock instead of waiting, and filename follow-ups run once after commit. --- src/documents/consumer.py | 15 +++- src/documents/tests/test_consumer.py | 127 +++++++++++++++++++++++++++ 2 files changed, 138 insertions(+), 4 deletions(-) diff --git a/src/documents/consumer.py b/src/documents/consumer.py index 5110fbd843..bb4261e3cd 100644 --- a/src/documents/consumer.py +++ b/src/documents/consumer.py @@ -26,6 +26,7 @@ from documents.file_handling import create_source_path_directory 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.file_moves import defer_file_moves from documents.loggers import LoggingMixin from documents.media_lock import media_lock from documents.models import Correspondent @@ -605,7 +606,14 @@ class ConsumerPlugin( # now that everything is done, we can start to store the document # in the system. This will be a transaction and reasonably fast. try: - with transaction.atomic(): + # File moves requested by saves in here run once, after + # commit and after media.lock is released (in production + # this nests under the consume task's scope). media.lock is + # taken before the transaction: the version path locks the + # root row with select_for_update, and workflows (email + # actions) and file placement below take media.lock again, + # re-entering instead of waiting while rows are locked. + with defer_file_moves(), media_lock(), transaction.atomic(): # store the document. if self.input_doc.root_document_id: # If this is a new version of an existing document, we need @@ -759,9 +767,8 @@ class ConsumerPlugin( document.archive_path, ) - # Don't save with the lock active. Saving will cause the file - # renaming logic to acquire the lock as well. - # This triggers things like file renaming + # Any filename follow-up from this save is deferred by + # the scope above until after commit. document.save(write_file_paths=True) if document.root_document_id: diff --git a/src/documents/tests/test_consumer.py b/src/documents/tests/test_consumer.py index d623070515..131b9edb64 100644 --- a/src/documents/tests/test_consumer.py +++ b/src/documents/tests/test_consumer.py @@ -27,6 +27,10 @@ from documents.models import Document from documents.models import DocumentType from documents.models import StoragePath from documents.models import Tag +from documents.models import Workflow +from documents.models import WorkflowAction +from documents.models import WorkflowActionEmail +from documents.models import WorkflowTrigger from documents.parsers import ParseError from documents.plugins.helpers import ProgressStatusOptions from documents.tasks import sanity_check @@ -38,6 +42,7 @@ from paperless_testing.factories import DocumentFactory from paperless_testing.factories import TagFactory from paperless_testing.factories import UserFactory from paperless_testing.fakes.progress import FakeProgressManager +from paperless_testing.locks import record_media_lock_acquisitions from paperless_testing.permissions import grant_object from paperless_testing.samples import MULTI_PAGE_DIGITAL_PDF from paperless_testing.samples import MULTI_PAGE_IMAGES_PDF @@ -221,6 +226,128 @@ class TestConsumer( shutil.copy(src, dst) return dst + @mock.patch("documents.consumer.load_classifier") + def test_consume_takes_media_lock_outside_transaction(self, m) -> None: + """ + GIVEN: + - A new document to consume + WHEN: + - It is consumed + THEN: + - media.lock is acquired + - media.lock is never first acquired inside a transaction + - The document's file is in place + """ + m.return_value = MagicMock() + + with record_media_lock_acquisitions() as acquisitions: + with self.get_consumer(self.get_test_file()) as consumer: + consumer.run() + + self.assertTrue(acquisitions) + self.assertFalse(any(a.in_user_transaction for a in acquisitions)) + document = Document.objects.first() + assert document is not None + self.assertIsFile(document.source_path) + + @mock.patch("documents.consumer.load_classifier") + def test_consume_version_takes_media_lock_outside_transaction(self, m) -> None: + """ + GIVEN: + - An existing root document + WHEN: + - A new version of it is consumed (the root row is locked with + select_for_update inside the consumer's transaction) + THEN: + - media.lock is never first acquired inside a transaction + - The root's modified is bumped + """ + m.return_value = MagicMock() + + with self.get_consumer(self.get_test_file()) as consumer: + consumer.run() + root_doc = Document.objects.first() + assert root_doc is not None + original_modified = timezone.now() - datetime.timedelta(days=1) + Document.objects.filter(pk=root_doc.pk).update(modified=original_modified) + + version_file = self.get_test_file2() + status = FakeProgressManager(version_file.name, None) + overrides = DocumentMetadataOverrides(version_label="v2") + doc = ConsumableDocument( + DocumentSource.ApiUpload, + original_file=version_file, + root_document_id=root_doc.pk, + ) + preflight = ConsumerPreflightPlugin( + doc, + overrides, + status, # type: ignore[arg-type] + self.dirs.scratch_dir, + "task-id", + ) + preflight.setup() + preflight.run() + consumer = ConsumerPlugin( + doc, + overrides, + status, # type: ignore[arg-type] + self.dirs.scratch_dir, + "task-id", + ) + consumer.setup() + + try: + with record_media_lock_acquisitions() as acquisitions: + consumer.run() + finally: + consumer.cleanup() + + self.assertTrue(acquisitions) + self.assertFalse(any(a.in_user_transaction for a in acquisitions)) + root_doc.refresh_from_db() + self.assertGreater(root_doc.modified, original_modified) + + @override_settings( + PAPERLESS_EMAIL_HOST="localhost", + EMAIL_ENABLED=True, + PAPERLESS_URL="http://localhost:8000", + ) + @mock.patch("documents.consumer.load_classifier") + @mock.patch("django.core.mail.message.EmailMessage.send", return_value=1) + def test_consumption_workflow_email_runs_under_held_lock(self, send, m) -> None: + """ + GIVEN: + - A document added workflow with an email action + WHEN: + - A document is consumed (the consumer holds media.lock while + workflows run) + THEN: + - The email action re-enters media.lock and sends without blocking + """ + m.return_value = MagicMock() + trigger = WorkflowTrigger.objects.create( + type=WorkflowTrigger.WorkflowTriggerType.DOCUMENT_ADDED, + ) + email = WorkflowActionEmail.objects.create( + subject="added", + body="body", + to="someone@example.com", + include_document=True, + ) + action = WorkflowAction.objects.create( + type=WorkflowAction.WorkflowActionType.EMAIL, + email=email, + ) + workflow = Workflow.objects.create(name="email on add") + workflow.triggers.add(trigger) + workflow.actions.add(action) + + with self.get_consumer(self.get_test_file()) as consumer: + consumer.run() + + send.assert_called_once() + @override_settings( FILENAME_FORMAT=None, TIME_ZONE="America/Chicago",