mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-11-04 22:09:24 +00:00
Fix: collect document file moves per scope and move once on exit
Each save of a document moved its files on the spot, so one logical change could move them several times, and moves could run inside another open transaction. defer_file_moves() opens a ContextVar-backed scope. Inside it, move requests are recorded and the outermost scope moves each document once on exit, or on commit if a transaction is still open. A failing move is logged and does not stop the others. Outside any scope moves stay immediate.
This commit is contained in:
1 parent
9a2b878859
commit
8f873604d3
2 files changed
+368
-4
No files matched your search
+113
-4
@@ -17,7 +17,11 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import shutil
|
||||
import threading
|
||||
from contextlib import contextmanager
|
||||
from contextvars import ContextVar
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import Final
|
||||
|
||||
from django.conf import settings
|
||||
@@ -35,6 +39,9 @@ from documents.media_lock import media_lock
|
||||
from documents.models import Document
|
||||
from documents.utils import compute_checksum
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
|
||||
logger = logging.getLogger("paperless.handlers")
|
||||
|
||||
MEDIA_LOCK_TIMEOUT_SECONDS: Final[int] = 300
|
||||
@@ -57,6 +64,47 @@ def in_user_transaction() -> bool:
|
||||
return any(not block._from_testcase for block in connection.atomic_blocks)
|
||||
|
||||
|
||||
class FileMoveCollector:
|
||||
"""Pending document pks for one file move scope."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._pending: set[int] = set()
|
||||
self._lock = threading.Lock()
|
||||
self.closed = False
|
||||
|
||||
def add(self, pk: int) -> bool:
|
||||
"""Record a pending move. Returns False if the collector is closed."""
|
||||
with self._lock:
|
||||
if self.closed:
|
||||
return False
|
||||
self._pending.add(pk)
|
||||
return True
|
||||
|
||||
def take_pending(self) -> list[int]:
|
||||
"""Remove and return the pending pks, sorted, leaving the collector open."""
|
||||
with self._lock:
|
||||
pending = sorted(self._pending)
|
||||
self._pending.clear()
|
||||
return pending
|
||||
|
||||
def close(self) -> list[int]:
|
||||
"""Close the collector and return what was still pending, sorted."""
|
||||
with self._lock:
|
||||
self.closed = True
|
||||
pending = sorted(self._pending)
|
||||
self._pending.clear()
|
||||
return pending
|
||||
|
||||
|
||||
# Only the scope context managers call set() / reset(). Everything else reads
|
||||
# it and mutates the collector, so adds made in a copied context (thread or
|
||||
# task hops) are visible to the scope that flushes.
|
||||
_active_collector: ContextVar[FileMoveCollector | None] = ContextVar(
|
||||
"paperless_file_move_collector",
|
||||
default=None,
|
||||
)
|
||||
|
||||
|
||||
def _path_matches_checksum(path: Path, checksum: str | None) -> bool:
|
||||
if checksum is None or not path.is_file():
|
||||
return False
|
||||
@@ -109,6 +157,62 @@ def move_files_for_document(pk: int) -> Document | None:
|
||||
return document
|
||||
|
||||
|
||||
def _flush(pks: list[int]) -> None:
|
||||
"""Move each document; one failure never stops the rest or raises."""
|
||||
for pk in pks:
|
||||
try:
|
||||
move_files_for_document(pk)
|
||||
except Exception:
|
||||
logger.exception(f"Unexpected error moving files of document {pk}")
|
||||
|
||||
|
||||
def _flush_after_commit(pks: list[int]) -> None:
|
||||
if not pks:
|
||||
return
|
||||
if in_user_transaction():
|
||||
transaction.on_commit(lambda: _flush(pks))
|
||||
else:
|
||||
_flush(pks)
|
||||
|
||||
|
||||
def _close_and_flush(collector: FileMoveCollector) -> None:
|
||||
_flush_after_commit(collector.close())
|
||||
|
||||
|
||||
@contextmanager
|
||||
def defer_file_moves() -> Iterator[None]:
|
||||
"""
|
||||
Collect file move requests and perform each once when the outermost scope
|
||||
exits, after its own writes have committed.
|
||||
"""
|
||||
current = _active_collector.get()
|
||||
if current is not None and not current.closed:
|
||||
yield
|
||||
return
|
||||
|
||||
collector = FileMoveCollector()
|
||||
token = _active_collector.set(collector)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
_active_collector.reset(token)
|
||||
_close_and_flush(collector)
|
||||
|
||||
|
||||
def flush_pending_file_moves() -> None:
|
||||
"""
|
||||
Perform the current scope's pending moves now, keeping the scope open.
|
||||
|
||||
When called from a task running eagerly inside a request, this flushes the
|
||||
request's collector early, which is harmless: the moves still run after
|
||||
commit and against fresh state.
|
||||
"""
|
||||
collector = _active_collector.get()
|
||||
if collector is None:
|
||||
return
|
||||
_flush_after_commit(collector.take_pending())
|
||||
|
||||
|
||||
def _validate_move(
|
||||
document: Document,
|
||||
old_path: Path,
|
||||
@@ -314,12 +418,17 @@ def _move_document_files(document: Document) -> None:
|
||||
|
||||
def request_file_move(document: Document) -> None:
|
||||
"""
|
||||
Ask for the document's files to follow its current committed state.
|
||||
Ask for the document's files to follow its committed state.
|
||||
|
||||
Moves immediately and copies only the final filename columns onto the
|
||||
caller's instance, so code that saves and then reads source_path keeps
|
||||
working without any other in-memory field being touched.
|
||||
Inside an open scope the request is recorded and performed when the scope
|
||||
ends. Otherwise the move happens now, and only the final filename columns
|
||||
are copied onto the caller's instance, so code that saves and then reads
|
||||
source_path keeps working without any other in-memory field being touched.
|
||||
"""
|
||||
collector = _active_collector.get()
|
||||
if collector is not None and collector.add(document.pk):
|
||||
return
|
||||
|
||||
moved = move_files_for_document(document.pk)
|
||||
if moved is not None:
|
||||
document.filename = moved.filename
|
||||
|
||||
@@ -6,9 +6,13 @@ import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
from django.db import transaction
|
||||
from django.utils import timezone
|
||||
from filelock import Timeout
|
||||
|
||||
from documents.file_moves import _active_collector
|
||||
from documents.file_moves import defer_file_moves
|
||||
from documents.file_moves import flush_pending_file_moves
|
||||
from documents.file_moves import move_files_for_document
|
||||
from documents.file_moves import request_file_move
|
||||
from documents.models import Document
|
||||
@@ -18,11 +22,16 @@ from paperless_testing.factories import TagFactory
|
||||
from paperless_testing.locks import record_media_lock_acquisitions
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
from contextlib import AbstractContextManager
|
||||
|
||||
from pytest_django.fixtures import Settings
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from paperless_testing.dirs import PaperlessDirs
|
||||
|
||||
OnCommitCapture = Callable[..., AbstractContextManager[list[Callable[[], object]]]]
|
||||
|
||||
CONTENT = b"original content"
|
||||
|
||||
|
||||
@@ -235,3 +244,249 @@ class TestReceiver:
|
||||
tag.documents.add(document)
|
||||
|
||||
request.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestDeferFileMoves:
|
||||
def test_moves_once_on_exit(
|
||||
self,
|
||||
document: Document,
|
||||
paperless_dirs: PaperlessDirs,
|
||||
) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- An open file move scope
|
||||
WHEN:
|
||||
- The document is saved with a new title inside the scope
|
||||
THEN:
|
||||
- Nothing moves while the scope is open
|
||||
- On exit the file and DB filename follow the final title
|
||||
"""
|
||||
with defer_file_moves():
|
||||
document.title = "renamed"
|
||||
document.save()
|
||||
assert (paperless_dirs.originals_dir / "original.pdf").is_file()
|
||||
assert Document.objects.get(pk=document.pk).filename == "original.pdf"
|
||||
|
||||
assert (paperless_dirs.originals_dir / "renamed.pdf").is_file()
|
||||
assert not (paperless_dirs.originals_dir / "original.pdf").exists()
|
||||
assert Document.objects.get(pk=document.pk).filename == "renamed.pdf"
|
||||
|
||||
def test_nested_scope_defers_to_outer(
|
||||
self,
|
||||
document: Document,
|
||||
paperless_dirs: PaperlessDirs,
|
||||
) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- A scope nested inside another
|
||||
WHEN:
|
||||
- The document is saved inside the inner scope and the inner scope exits
|
||||
THEN:
|
||||
- Nothing moves until the outer scope exits
|
||||
"""
|
||||
with defer_file_moves():
|
||||
with defer_file_moves():
|
||||
document.title = "renamed"
|
||||
document.save()
|
||||
assert (paperless_dirs.originals_dir / "original.pdf").is_file()
|
||||
|
||||
assert (paperless_dirs.originals_dir / "renamed.pdf").is_file()
|
||||
|
||||
def test_closed_collector_moves_immediately(
|
||||
self,
|
||||
document: Document,
|
||||
paperless_dirs: PaperlessDirs,
|
||||
) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- A collector that has already been closed (work outliving its scope)
|
||||
WHEN:
|
||||
- A save asks for a move while that collector is still current
|
||||
THEN:
|
||||
- The move happens immediately instead of being lost
|
||||
"""
|
||||
with defer_file_moves():
|
||||
collector = _active_collector.get()
|
||||
assert collector is not None
|
||||
token = _active_collector.set(collector)
|
||||
try:
|
||||
document.title = "renamed"
|
||||
document.save()
|
||||
finally:
|
||||
_active_collector.reset(token)
|
||||
|
||||
assert (paperless_dirs.originals_dir / "renamed.pdf").is_file()
|
||||
|
||||
def test_scope_closing_inside_atomic_flushes_on_commit(
|
||||
self,
|
||||
document: Document,
|
||||
paperless_dirs: PaperlessDirs,
|
||||
django_capture_on_commit_callbacks: OnCommitCapture,
|
||||
) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- A scope opened and closed inside an atomic block
|
||||
WHEN:
|
||||
- The atomic block commits
|
||||
THEN:
|
||||
- The move runs only once the commit callbacks run
|
||||
"""
|
||||
with django_capture_on_commit_callbacks(execute=True):
|
||||
with transaction.atomic(), defer_file_moves():
|
||||
document.title = "renamed"
|
||||
document.save()
|
||||
assert (paperless_dirs.originals_dir / "original.pdf").is_file()
|
||||
|
||||
assert (paperless_dirs.originals_dir / "renamed.pdf").is_file()
|
||||
|
||||
def test_scope_closing_inside_rolled_back_atomic_moves_nothing(
|
||||
self,
|
||||
document: Document,
|
||||
paperless_dirs: PaperlessDirs,
|
||||
django_capture_on_commit_callbacks: OnCommitCapture,
|
||||
) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- A scope opened and closed inside an atomic block
|
||||
WHEN:
|
||||
- The atomic block rolls back
|
||||
THEN:
|
||||
- The title change is gone and the file is not moved
|
||||
"""
|
||||
|
||||
class Rollback(Exception):
|
||||
pass
|
||||
|
||||
with django_capture_on_commit_callbacks(execute=True):
|
||||
with pytest.raises(Rollback):
|
||||
with transaction.atomic(), defer_file_moves():
|
||||
document.title = "renamed"
|
||||
document.save()
|
||||
raise Rollback
|
||||
|
||||
assert (paperless_dirs.originals_dir / "original.pdf").is_file()
|
||||
assert not (paperless_dirs.originals_dir / "renamed.pdf").exists()
|
||||
assert Document.objects.get(pk=document.pk).title == "original"
|
||||
|
||||
def test_flush_pending_moves_early(
|
||||
self,
|
||||
document: Document,
|
||||
paperless_dirs: PaperlessDirs,
|
||||
) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- An open scope with a pending move
|
||||
WHEN:
|
||||
- flush_pending_file_moves is called inside the scope
|
||||
THEN:
|
||||
- The move happens before the scope exits
|
||||
"""
|
||||
with defer_file_moves():
|
||||
document.title = "renamed"
|
||||
document.save()
|
||||
flush_pending_file_moves()
|
||||
assert (paperless_dirs.originals_dir / "renamed.pdf").is_file()
|
||||
|
||||
def test_document_without_filename_in_scope_is_noop(
|
||||
self,
|
||||
paperless_dirs: PaperlessDirs,
|
||||
title_format: None,
|
||||
) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- A document still being consumed (no filename yet)
|
||||
WHEN:
|
||||
- It is saved inside a scope and the scope exits
|
||||
THEN:
|
||||
- Nothing is written, media.lock is never taken, nothing raises
|
||||
"""
|
||||
doc = DocumentFactory(filename=None, mime_type="application/pdf")
|
||||
|
||||
with record_media_lock_acquisitions() as acquisitions:
|
||||
with defer_file_moves():
|
||||
doc.title = "renamed"
|
||||
doc.save()
|
||||
|
||||
assert acquisitions == []
|
||||
doc.refresh_from_db()
|
||||
assert doc.filename is None
|
||||
|
||||
def test_mover_keeps_metadata_changed_after_save(
|
||||
self,
|
||||
document: Document,
|
||||
paperless_dirs: PaperlessDirs,
|
||||
settings: Settings,
|
||||
) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- A filename format using correspondent and title
|
||||
- A scope in which the title is saved
|
||||
WHEN:
|
||||
- Another writer changes the correspondent before the scope exits
|
||||
THEN:
|
||||
- The correspondent change survives the flush
|
||||
- The final path reflects both changes
|
||||
"""
|
||||
settings.FILENAME_FORMAT = "{correspondent}/{title}"
|
||||
correspondent = CorrespondentFactory(name="acme")
|
||||
|
||||
with defer_file_moves():
|
||||
document.title = "renamed"
|
||||
document.save()
|
||||
Document.objects.filter(pk=document.pk).update(correspondent=correspondent)
|
||||
|
||||
document.refresh_from_db()
|
||||
assert document.correspondent == correspondent
|
||||
assert document.filename == "acme/renamed.pdf"
|
||||
assert (paperless_dirs.originals_dir / "acme" / "renamed.pdf").is_file()
|
||||
|
||||
def test_one_failure_does_not_stop_the_flush(
|
||||
self,
|
||||
paperless_dirs: PaperlessDirs,
|
||||
title_format: None,
|
||||
mocker: MockerFixture,
|
||||
) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- Two documents saved in one scope
|
||||
- Moving the first raises an unexpected error
|
||||
WHEN:
|
||||
- The scope exits
|
||||
THEN:
|
||||
- The scope does not raise
|
||||
- The second document is still moved
|
||||
"""
|
||||
(paperless_dirs.originals_dir / "a.pdf").write_bytes(CONTENT)
|
||||
first = DocumentFactory(
|
||||
filename="a.pdf",
|
||||
title="a",
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
(paperless_dirs.originals_dir / "b.pdf").write_bytes(CONTENT)
|
||||
second = DocumentFactory(
|
||||
filename="b.pdf",
|
||||
title="b",
|
||||
mime_type="application/pdf",
|
||||
)
|
||||
|
||||
real_move = move_files_for_document
|
||||
|
||||
def failing_for_first(pk: int) -> Document | None:
|
||||
if pk == first.pk:
|
||||
raise RuntimeError("boom")
|
||||
return real_move(pk)
|
||||
|
||||
mocker.patch(
|
||||
"documents.file_moves.move_files_for_document",
|
||||
side_effect=failing_for_first,
|
||||
)
|
||||
|
||||
with defer_file_moves():
|
||||
first.title = "a2"
|
||||
first.save()
|
||||
second.title = "b2"
|
||||
second.save()
|
||||
|
||||
assert (paperless_dirs.originals_dir / "b2.pdf").is_file()
|
||||
assert (paperless_dirs.originals_dir / "a.pdf").is_file()
|
||||
Reference in new issue
Block a user