mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-11-04 22:09:24 +00:00
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.
This commit is contained in:
1 parent
5bcde3205f
commit
1f3b6db58d
3 files changed
+106
-1
No files matched your search
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
+20
-1
@@ -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.
|
||||
|
||||
Reference in new issue
Block a user