diff --git a/docs/api.md b/docs/api.md index cf010e2f95..596e4bc375 100644 --- a/docs/api.md +++ b/docs/api.md @@ -290,9 +290,14 @@ a json payload of the format: Document bulk operations, including the dedicated document action endpoints, return `{"result":"OK","task_id":"…"}` when queued. Use `/api/tasks/?task_id={uuid}` -to check the task outcome. Actions that create or reprocess files may start further -consumption or reprocessing tasks and completion of the bulk task means those tasks -were scheduled, not that their file processing has finished. +to check the task outcome. Reprocessing operations remain started until all of their +individual document tasks have finished. Their results include `succeeded_count` and +`failed_count`; any failed or revoked child makes the operation fail, while successful +changes remain applied. A scheduling failure is also reported, after any tasks already +published have finished. Child tasks belong to the initiating user. + +Other actions that create files may start further consumption tasks. Completion of +those bulk tasks means consumption was scheduled, not that file processing has finished. ```json { diff --git a/src/documents/bulk_edit.py b/src/documents/bulk_edit.py index 762aa6512b..41bc5931ce 100644 --- a/src/documents/bulk_edit.py +++ b/src/documents/bulk_edit.py @@ -432,7 +432,12 @@ def delete(doc_ids: list[int]) -> Literal["OK"]: return "OK" -def reprocess(doc_ids: list[int], *, remote_ocr: bool = False) -> Literal["OK"]: +def reprocess( + doc_ids: list[int], + *, + remote_ocr: bool = False, + parent_task_id: str | None = None, +) -> Literal["OK"]: """ Re-run parsing for the given documents. @@ -457,6 +462,16 @@ def reprocess(doc_ids: list[int], *, remote_ocr: bool = False) -> Literal["OK"]: .values_list("source_id", flat=True) .distinct() ) + if parent_task_id is not None: + from documents.task_tracking import queue_reprocessing_tasks + + queue_reprocessing_tasks( + parent_task_id, + list(source_ids), + remote_ocr=remote_ocr, + ) + return "OK" + for source_id in source_ids: update_document_content_maybe_archive_file.apply_async( kwargs={"document_id": source_id, "remote_ocr": remote_ocr}, diff --git a/src/documents/bulk_operations.py b/src/documents/bulk_operations.py index 57cc55f862..a1965c4cee 100644 --- a/src/documents/bulk_operations.py +++ b/src/documents/bulk_operations.py @@ -89,8 +89,9 @@ def snapshot_field(doc_ids: list[int], field: str) -> dict[int, Any]: return values -@shared_task +@shared_task(bind=True) def execute_bulk_operation( + self, operation: str, document_ids: list[int], parameters: dict, @@ -114,6 +115,8 @@ def execute_bulk_operation( if operation not in MODIFIED_FIELDS: execution_parameters = parameters.copy() + if operation == "reprocess" and self.request.id is not None: + execution_parameters["parent_task_id"] = self.request.id if operation in USER_OPERATIONS: execution_parameters["user"] = user if operation in TRIGGER_SOURCE_OPERATIONS: diff --git a/src/documents/migrations/0031_task_parent_tracking.py b/src/documents/migrations/0031_task_parent_tracking.py new file mode 100644 index 0000000000..96b7f4dc60 --- /dev/null +++ b/src/documents/migrations/0031_task_parent_tracking.py @@ -0,0 +1,30 @@ +# Generated by Django 5.2.18 on 2026-11-14 19:59 + +import django.db.models.deletion +from django.db import migrations +from django.db import models + + +class Migration(migrations.Migration): + dependencies = [ + ("documents", "0030_note_format"), + ] + + operations = [ + migrations.AddField( + model_name="paperlesstask", + name="dispatched", + field=models.BooleanField(default=False), + ), + migrations.AddField( + model_name="paperlesstask", + name="parent", + field=models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="children", + to="documents.paperlesstask", + ), + ), + ] diff --git a/src/documents/models.py b/src/documents/models.py index 12444a8de5..f5ab857dc1 100644 --- a/src/documents/models.py +++ b/src/documents/models.py @@ -970,6 +970,16 @@ class PaperlessTask(ModelWithOwner): help_text=_("Structured result data from task execution"), ) + parent = models.ForeignKey( + "self", + null=True, + blank=True, + on_delete=models.SET_NULL, + related_name="children", + ) + + dispatched = models.BooleanField(default=False) + # Acknowledgment acknowledged = models.BooleanField( default=False, diff --git a/src/documents/signals/handlers.py b/src/documents/signals/handlers.py index cb5c376b14..2f044afd97 100644 --- a/src/documents/signals/handlers.py +++ b/src/documents/signals/handlers.py @@ -56,6 +56,8 @@ from documents.models import WorkflowRun from documents.models import WorkflowTrigger from documents.permissions import restrict_queryset_to_visible from documents.plugins.helpers import DocumentsStatusManager +from documents.task_tracking import finish_dispatch +from documents.task_tracking import update_parent from documents.templating.utils import convert_format_str_to_template_format from documents.utils import compute_checksum from documents.utils import copy_file_with_basic_stats @@ -1309,6 +1311,10 @@ def task_postrun_handler( except PaperlessTask.DoesNotExist: return + if task_instance.children.exists(): + finish_dispatch(task_instance, retval if isinstance(retval, dict) else {}) + return + task_instance.status = new_status task_instance.date_done = now changed_fields = ["status", "date_done"] @@ -1332,6 +1338,7 @@ def task_postrun_handler( changed_fields.append("status") task_instance.save(update_fields=changed_fields) + update_parent(task_instance) if task and task.name == BULK_OPERATION_TASK_NAME: _send_bulk_operation_status(task_instance) except Exception: # pragma: no cover @@ -1388,6 +1395,10 @@ def task_failure_handler( } task_qs = PaperlessTask.objects.filter(task_id=task_id) + parent = task_qs.first() + if parent is not None and parent.children.exists(): + finish_dispatch(parent, result_data) + return task_instance = task_qs.values("date_started", "date_created").first() if task_instance: date_started = task_instance["date_started"] @@ -1399,6 +1410,7 @@ def task_failure_handler( date_started - date_created ).total_seconds() task_qs.update(**update_fields) + update_parent(task_qs.get()) if sender and sender.name == BULK_OPERATION_TASK_NAME: _send_bulk_operation_status(task_qs.first()) except Exception: # pragma: no cover @@ -1445,6 +1457,8 @@ def task_revoked_handler( "reason": (reason + " Check worker logs for details."), }, ) + if task_instance := task_qs.first(): + update_parent(task_instance) if sender and sender.name == BULK_OPERATION_TASK_NAME: _send_bulk_operation_status(task_qs.first()) except Exception: # pragma: no cover diff --git a/src/documents/task_tracking.py b/src/documents/task_tracking.py new file mode 100644 index 0000000000..d8336f6ed4 --- /dev/null +++ b/src/documents/task_tracking.py @@ -0,0 +1,111 @@ +"""Completion tracking for operations which dispatch document processing tasks.""" + +import uuid + +from django.db import transaction +from django.utils import timezone + +from documents.models import PaperlessTask + + +def queue_reprocessing_tasks( + parent_task_id: str, + document_ids: list[int], + *, + remote_ocr: bool, +) -> None: + from documents.tasks import update_document_content_maybe_archive_file + + parent = PaperlessTask.objects.get(task_id=parent_task_id) + children = [ + PaperlessTask( + task_id=str(uuid.uuid4()), + task_type=PaperlessTask.TaskType.REPROCESS_DOCUMENT, + trigger_source=parent.trigger_source, + owner_id=parent.owner_id, + parent=parent, + input_data={"document_id": document_id}, + ) + for document_id in document_ids + ] + # Register the complete set before any can finish + PaperlessTask.objects.bulk_create(children) + for index, child in enumerate(children): + try: + update_document_content_maybe_archive_file.apply_async( + task_id=child.task_id, + kwargs={ + "document_id": child.input_data["document_id"], + "remote_ocr": remote_ocr, + }, + headers={"trigger_source": parent.trigger_source}, + ) + except Exception: + PaperlessTask.objects.filter( + pk__in=[task.pk for task in children[index:]], + ).update( + status=PaperlessTask.Status.FAILURE, + date_done=timezone.now(), + result_data={"error_message": "Could not publish reprocessing task."}, + ) + raise + + +def finish_dispatch(task: PaperlessTask, result_data: dict) -> None: + """Record the scheduling outcome without completing an active parent.""" + with transaction.atomic(): + parent = PaperlessTask.objects.select_for_update().get(pk=task.pk) + if parent.is_complete: + return + parent.dispatched = True + parent.result_data = result_data + parent.save(update_fields=["dispatch_complete", "result_data"]) + _finalize_parent(parent) + + +def update_parent(task: PaperlessTask) -> None: + if task.parent_id is None: + return + with transaction.atomic(): + parent = PaperlessTask.objects.select_for_update().get(pk=task.parent_id) + _finalize_parent(parent) + + +def _finalize_parent(parent: PaperlessTask) -> None: + if parent.is_complete or not parent.dispatched: + return + statuses = list(parent.children.values_list("status", flat=True)) + if any(status not in PaperlessTask.COMPLETE_STATUSES for status in statuses): + return + + succeeded = statuses.count(PaperlessTask.Status.SUCCESS) + failed = len(statuses) - succeeded + result_data = parent.result_data or {} + if failed and not result_data.get("error_message"): + result_data["error_message"] = ( + f"{succeeded} document(s) reprocessed; {failed} failed or were revoked. " + "See child tasks for details." + ) + parent.status = ( + PaperlessTask.Status.FAILURE + if result_data.get("error_message") + else PaperlessTask.Status.SUCCESS + ) + parent.result_data = { + **parent.input_data, + **result_data, + "succeeded_count": succeeded, + "failed_count": failed, + } + parent.date_done = timezone.now() + if parent.date_started: + parent.duration_seconds = ( + parent.date_done - parent.date_started + ).total_seconds() + parent.wait_time_seconds = ( + parent.date_started - parent.date_created + ).total_seconds() + parent.save() + from documents.signals.handlers import _send_bulk_operation_status + + transaction.on_commit(lambda: _send_bulk_operation_status(parent)) diff --git a/src/documents/tests/test_task_tracking.py b/src/documents/tests/test_task_tracking.py new file mode 100644 index 0000000000..1299895982 --- /dev/null +++ b/src/documents/tests/test_task_tracking.py @@ -0,0 +1,176 @@ +import uuid +from unittest import mock + +import pytest + +from documents.bulk_operations import execute_bulk_operation +from documents.models import PaperlessTask +from documents.signals.handlers import task_failure_handler +from documents.signals.handlers import task_postrun_handler +from documents.signals.handlers import task_revoked_handler +from documents.tasks import update_document_content_maybe_archive_file +from paperless_testing.factories import DocumentFactory +from paperless_testing.factories import UserFactory + + +@pytest.fixture +def operation(paperless_dirs): + user = UserFactory(superuser=True) + documents = DocumentFactory.create_batch(3) + parent = PaperlessTask.objects.create( + task_id=str(uuid.uuid4()), + task_type=PaperlessTask.TaskType.BULK_UPDATE, + owner=user, + trigger_source=PaperlessTask.TriggerSource.WEB_UI, + input_data={"operation": "reprocess", "document_count": 3}, + ) + kwargs = { + "operation": "reprocess", + "document_ids": [document.pk for document in documents], + "parameters": {"remote_ocr": True}, + "user_id": user.pk, + } + return parent, kwargs + + +def complete(child, status="SUCCESS"): + if status == "FAILURE": + task_failure_handler( + sender=update_document_content_maybe_archive_file, + task_id=child.task_id, + exception=RuntimeError("Parser failed"), + ) + elif status == "REVOKED": + task_revoked_handler( + sender=update_document_content_maybe_archive_file, + request=mock.Mock(id=child.task_id), + ) + else: + task_postrun_handler( + task=update_document_content_maybe_archive_file, + task_id=child.task_id, + state=status, + ) + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.parametrize("outcome", ["SUCCESS", "FAILURE", "REVOKED"]) +def test_parent_waits_for_all_children(operation, outcome): + parent, kwargs = operation + with ( + mock.patch( + "documents.tasks.update_document_content_maybe_archive_file.apply_async", + ) as publish, + mock.patch("documents.signals.handlers._send_bulk_operation_status") as notify, + ): + execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True) + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.STARTED + assert parent.date_done is None + children = list(parent.children.order_by("pk")) + assert len(children) == 3 + assert publish.call_count == 3 + assert all(child.owner_id == parent.owner_id for child in children) + assert all(child.trigger_source == parent.trigger_source for child in children) + assert all( + call.kwargs["kwargs"]["remote_ocr"] for call in publish.call_args_list + ) + complete(children[0], outcome) + complete(children[1]) + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.STARTED + notify.assert_not_called() + complete(children[2]) + parent.refresh_from_db() + assert parent.status == ("success" if outcome == "SUCCESS" else "failure") + assert parent.result_data["succeeded_count"] == ( + 3 if outcome == "SUCCESS" else 2 + ) + assert parent.result_data["failed_count"] == (0 if outcome == "SUCCESS" else 1) + assert parent.date_done is not None + assert parent.duration_seconds >= 0 + notify.assert_called_once() + complete(children[2]) + notify.assert_called_once() + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.parametrize("outcome", ["SUCCESS", "FAILURE"]) +def test_fast_children_wait_for_dispatch_to_finish(operation, outcome): + parent, kwargs = operation + + def publish(*, task_id, **kwargs): + # Every child is registered before the first publication. + assert parent.children.count() == 3 + complete(PaperlessTask.objects.get(task_id=task_id), outcome) + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.STARTED + assert not parent.dispatch_complete + + with ( + mock.patch( + "documents.tasks.update_document_content_maybe_archive_file.apply_async", + side_effect=publish, + ), + mock.patch("documents.signals.handlers._send_bulk_operation_status") as notify, + ): + execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True) + parent.refresh_from_db() + assert parent.status == ("success" if outcome == "SUCCESS" else "failure") + notify.assert_called_once() + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.parametrize("published_count", [0, 1]) +def test_publish_failure_waits_for_published_children(operation, published_count): + parent, kwargs = operation + with ( + mock.patch( + "documents.tasks.update_document_content_maybe_archive_file.apply_async", + side_effect=[None] * published_count + [RuntimeError("Broker unavailable")], + ), + mock.patch("documents.signals.handlers._send_bulk_operation_status") as notify, + ): + execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=False) + parent.refresh_from_db() + if published_count: + assert parent.status == PaperlessTask.Status.STARTED + notify.assert_not_called() + complete(parent.children.get(status=PaperlessTask.Status.PENDING)) + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.FAILURE + assert parent.result_data["error_message"] == "Broker unavailable" + assert parent.result_data["succeeded_count"] == published_count + assert parent.result_data["failed_count"] == 3 - published_count + notify.assert_called_once() + + +@pytest.mark.django_db(transaction=True) +def test_empty_selection_completes_without_children(operation): + parent, kwargs = operation + kwargs["document_ids"] = [] + execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True) + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.SUCCESS + assert not parent.children.exists() + + +@pytest.mark.django_db +def test_client_cannot_supply_parent_task_id(operation): + from rest_framework.test import APIClient + + parent, kwargs = operation + client = APIClient() + client.force_authenticate(parent.owner) + with mock.patch("documents.views.execute_bulk_operation.apply_async") as publish: + response = client.post( + "/api/documents/bulk_edit/", + { + "documents": kwargs["document_ids"], + "method": "set_correspondent", + "parameters": {"correspondent": None, "parent_task_id": parent.task_id}, + }, + format="json", + ) + assert response.status_code == 400 + publish.assert_not_called() diff --git a/src/documents/views.py b/src/documents/views.py index 042b8e2c47..8c95a0809d 100644 --- a/src/documents/views.py +++ b/src/documents/views.py @@ -2976,8 +2976,9 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin): parser_classes = (parsers.JSONParser,) def _queue_bulk_operation(self, method, documents, parameters, from_webui): - if "update_callback" in parameters: - raise ValueError("Unsupported bulk operation parameter: update_callback") + for parameter in ("update_callback", "parent_task_id"): + if parameter in parameters: + raise ValueError(f"Unsupported bulk operation parameter: {parameter}") trigger_source = ( PaperlessTask.TriggerSource.WEB_UI if from_webui