From 1f3b6db58d2fe1c4b67106e578a8b6cf7ca9901a Mon Sep 17 00:00:00 2001 From: stumpylog <797416+stumpylog@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:18:16 -0800 Subject: [PATCH] Fix: move document files once per Celery task Tasks such as bulk_update_documents moved a document's files once for its manual post_save and again for every save its workflows made, each time taking the global media lock. All tasks now run through FileMoveScopedTask, which opens a file move scope around the task body, including when the task raises. bulk_update_documents flushes after each document so long bulk edits still make steady progress. --- src/documents/tasks.py | 4 ++ src/documents/tests/test_file_moves.py | 82 ++++++++++++++++++++++++++ src/paperless/celery.py | 21 ++++++- 3 files changed, 106 insertions(+), 1 deletion(-) diff --git a/src/documents/tasks.py b/src/documents/tasks.py index ac6db4efb2..4ddc84a3d3 100644 --- a/src/documents/tasks.py +++ b/src/documents/tasks.py @@ -310,6 +310,7 @@ def sanity_check(*, raise_on_error: bool = True) -> str: @shared_task def bulk_update_documents(document_ids) -> None: + from documents.file_moves import flush_pending_file_moves from documents.search import get_backend document_ids = list(document_ids) @@ -330,6 +331,9 @@ def bulk_update_documents(document_ids) -> None: skip_ai_index=True, # bulk path calls update_llm_index once below ) post_save.send(Document, instance=doc, created=False) + # Move this document's files now rather than at the end of the task, + # so a long bulk edit makes steady progress. + flush_pending_file_moves() with get_backend().batch_update() as batch: batch.add_or_update_ids(document_ids) diff --git a/src/documents/tests/test_file_moves.py b/src/documents/tests/test_file_moves.py index a4dea358e4..452c5fce87 100644 --- a/src/documents/tests/test_file_moves.py +++ b/src/documents/tests/test_file_moves.py @@ -3,6 +3,7 @@ from __future__ import annotations import datetime import hashlib import logging +from pathlib import Path from typing import TYPE_CHECKING import pytest @@ -23,6 +24,8 @@ from documents.file_moves import move_files_for_document from documents.file_moves import request_file_move from documents.models import CustomField from documents.models import Document +from documents.tasks import bulk_update_documents +from paperless.celery import app as celery_app from paperless.middleware import FileMoveMiddleware from paperless_testing.factories import CorrespondentFactory from paperless_testing.factories import CustomFieldFactory @@ -64,6 +67,16 @@ def document(paperless_dirs: PaperlessDirs, title_format: None) -> Document: ) +@celery_app.task(name="documents.tests.test_file_moves.rename_then_fail") +def rename_then_fail(pk: int, originals_dir: str) -> None: + """Test-only task: rename the document, check nothing moved yet, then fail.""" + doc = Document.objects.get(pk=pk) + doc.title = "renamed" + doc.save() + assert (Path(originals_dir) / "original.pdf").is_file() + raise RuntimeError("task failed after saving") + + @pytest.mark.django_db class TestMoveFilesForDocument: def test_moves_using_fresh_db_state( @@ -751,3 +764,72 @@ class TestFileMovesThroughApi: document.refresh_from_db() assert document.filename == "cherry/original.pdf" assert (paperless_dirs.originals_dir / "cherry" / "original.pdf").is_file() + + +@pytest.mark.django_db +class TestCeleryTaskScope: + def test_task_class_opens_scope(self) -> None: + """ + GIVEN: + - The paperless Celery app + WHEN: + - A registered task is inspected + THEN: + - It runs through FileMoveScopedTask + """ + from paperless.celery import FileMoveScopedTask + + assert isinstance(bulk_update_documents, FileMoveScopedTask) + + def test_failing_task_still_moves_committed_change( + self, + document: Document, + paperless_dirs: PaperlessDirs, + ) -> None: + """ + GIVEN: + - A task that renames a document, then raises + WHEN: + - The task is applied + THEN: + - The task fails with its own error + - Nothing moved while the task body ran + - The committed rename is moved when the task ends + """ + result = rename_then_fail.apply( + args=[document.pk, str(paperless_dirs.originals_dir)], + ) + + assert result.failed() + assert isinstance(result.result, RuntimeError) + assert (paperless_dirs.originals_dir / "renamed.pdf").is_file() + + def test_bulk_update_moves_each_document( + self, + document: Document, + paperless_dirs: PaperlessDirs, + mocker: MockerFixture, + ) -> None: + """ + GIVEN: + - Two documents whose titles changed in the DB without signals + WHEN: + - bulk_update_documents runs for both + THEN: + - Each document's file follows its new title + """ + mocker.patch("documents.search.get_backend") + (paperless_dirs.originals_dir / "second.pdf").write_bytes(CONTENT) + second = DocumentFactory( + title="second", + filename="second.pdf", + mime_type="application/pdf", + checksum=hashlib.sha256(CONTENT).hexdigest(), + ) + Document.objects.filter(pk=document.pk).update(title="first-renamed") + Document.objects.filter(pk=second.pk).update(title="second-renamed") + + bulk_update_documents.apply(args=[[document.pk, second.pk]]) + + assert (paperless_dirs.originals_dir / "first-renamed.pdf").is_file() + assert (paperless_dirs.originals_dir / "second-renamed.pdf").is_file() diff --git a/src/paperless/celery.py b/src/paperless/celery.py index e644a71b24..dab9217a07 100644 --- a/src/paperless/celery.py +++ b/src/paperless/celery.py @@ -1,6 +1,7 @@ import os from celery import Celery +from celery import Task from celery.signals import worker_process_init from kombu.serialization import register @@ -26,7 +27,25 @@ register( content_encoding="binary", ) -app = Celery("paperless") + +class FileMoveScopedTask(Task): + """ + Run every task inside a document file move scope, so each document a task + changes is moved once when the task ends, after its writes have committed. + The scope also flushes when the task raises, for writes that committed + before the failure. + """ + + def __call__(self, *args, **kwargs): + # Function-local: this module is imported by paperless/__init__.py + # before Django apps are loaded. + from documents.file_moves import defer_file_moves + + with defer_file_moves(): + return super().__call__(*args, **kwargs) + + +app = Celery("paperless", task_cls=FileMoveScopedTask) # Using a string here means the worker doesn't have to serialize # the configuration object to child processes.