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",