mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-11-10 08:21:01 +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
6515136cdf
commit
b77e74ebe2
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
|
@shared_task
|
||||||
def bulk_update_documents(document_ids) -> None:
|
def bulk_update_documents(document_ids) -> None:
|
||||||
|
from documents.file_moves import flush_pending_file_moves
|
||||||
from documents.search import get_backend
|
from documents.search import get_backend
|
||||||
|
|
||||||
document_ids = list(document_ids)
|
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
|
skip_ai_index=True, # bulk path calls update_llm_index once below
|
||||||
)
|
)
|
||||||
post_save.send(Document, instance=doc, created=False)
|
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:
|
with get_backend().batch_update() as batch:
|
||||||
batch.add_or_update_ids(document_ids)
|
batch.add_or_update_ids(document_ids)
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ from __future__ import annotations
|
|||||||
import datetime
|
import datetime
|
||||||
import hashlib
|
import hashlib
|
||||||
import logging
|
import logging
|
||||||
|
from pathlib import Path
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
import pytest
|
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.file_moves import request_file_move
|
||||||
from documents.models import CustomField
|
from documents.models import CustomField
|
||||||
from documents.models import Document
|
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.middleware import FileMoveMiddleware
|
||||||
from paperless_testing.factories import CorrespondentFactory
|
from paperless_testing.factories import CorrespondentFactory
|
||||||
from paperless_testing.factories import CustomFieldFactory
|
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
|
@pytest.mark.django_db
|
||||||
class TestMoveFilesForDocument:
|
class TestMoveFilesForDocument:
|
||||||
def test_moves_using_fresh_db_state(
|
def test_moves_using_fresh_db_state(
|
||||||
@@ -751,3 +764,72 @@ class TestFileMovesThroughApi:
|
|||||||
document.refresh_from_db()
|
document.refresh_from_db()
|
||||||
assert document.filename == "cherry/original.pdf"
|
assert document.filename == "cherry/original.pdf"
|
||||||
assert (paperless_dirs.originals_dir / "cherry" / "original.pdf").is_file()
|
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
|
import os
|
||||||
|
|
||||||
from celery import Celery
|
from celery import Celery
|
||||||
|
from celery import Task
|
||||||
from celery.signals import worker_process_init
|
from celery.signals import worker_process_init
|
||||||
from kombu.serialization import register
|
from kombu.serialization import register
|
||||||
|
|
||||||
@@ -26,7 +27,25 @@ register(
|
|||||||
content_encoding="binary",
|
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
|
# Using a string here means the worker doesn't have to serialize
|
||||||
# the configuration object to child processes.
|
# the configuration object to child processes.
|
||||||
|
|||||||
Reference in new issue
Block a user