From 8f873604d3af9e7487189813a5137715be679792 Mon Sep 17 00:00:00 2001 From: stumpylog <797416+stumpylog@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:10:02 -0800 Subject: [PATCH] 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. --- src/documents/file_moves.py | 117 +++++++++++- src/documents/tests/test_file_moves.py | 255 +++++++++++++++++++++++++ 2 files changed, 368 insertions(+), 4 deletions(-) diff --git a/src/documents/file_moves.py b/src/documents/file_moves.py index a3f3cdb037..f7718f2152 100644 --- a/src/documents/file_moves.py +++ b/src/documents/file_moves.py @@ -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 diff --git a/src/documents/tests/test_file_moves.py b/src/documents/tests/test_file_moves.py index f198356704..39317572f3 100644 --- a/src/documents/tests/test_file_moves.py +++ b/src/documents/tests/test_file_moves.py @@ -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()