Raise proper Pulp errors from failing tasks

CHANGES/+unsafe-task-exceptions.bugfix

@@ -0,0 +1 @@
+Task failures that previously surfaced as unhandled Python exceptions (deleting an object still referenced by others, replication errors, and failed subtasks) now raise proper Pulp errors, so their messages are preserved and reported clearly.

pulpcore/app/tasks/base.py

@@ -3,10 +3,12 @@
 
 from asgiref.sync import sync_to_async
 from django.db import transaction
+from django.db.models.deletion import ProtectedError
 
 from pulpcore.app.apps import get_plugin_config
 from pulpcore.app.loggers import deprecation_logger
 from pulpcore.app.models import CreatedResource
+from pulpcore.exceptions import ProtectedResourceError
 from pulpcore.plugin.models import MasterModel
 
 log = getLogger(__name__)
@@ -122,7 +124,10 @@ def general_delete(instance_id, app_label, serializer_name, **kwargs):
         return dict(output)
     if isinstance(instance, MasterModel):
         instance = instance.cast()
-    output.update(instance.delete()[1])
+    try:
+        output.update(instance.delete()[1])
+    except ProtectedError as e:
+        raise ProtectedResourceError(details=str(e))
     return dict(output)
 
 
@@ -159,10 +164,13 @@ def general_multi_delete(instance_ids, **kwargs):
         if isinstance(instance, MasterModel):
             instance = instance.cast()
         instances.append(instance)
-    with transaction.atomic():
-        for instance in instances:
-            for model_label, count in instance.delete()[1].items():
-                counts[model_label] += count
+    try:
+        with transaction.atomic():
+            for instance in instances:
+                for model_label, count in instance.delete()[1].items():
+                    counts[model_label] += count
+    except ProtectedError as e:
+        raise ProtectedResourceError(details=str(e))
     output.update(counts)
     return dict(output)
 
@@ -209,5 +217,8 @@ async def ageneral_delete(instance_id, app_label, serializer_name, **kwargs):
         return dict(output)
     if isinstance(instance, MasterModel):
         instance = await instance.acast()
-    output.update((await instance.adelete())[1])
+    try:
+        output.update((await instance.adelete())[1])
+    except ProtectedError as e:
+        raise ProtectedResourceError(details=str(e))
     return dict(output)

pulpcore/app/tasks/replica.py

@@ -3,6 +3,7 @@
 import sys
 from tempfile import NamedTemporaryFile
 
+import requests
 from django.db import transaction
 from django.db.models import Min
 from pulp_glue.common import __version__ as pulp_glue_version
@@ -13,7 +14,7 @@
 from pulpcore.app.models import Distribution, Repository, Task, TaskGroup, UpstreamPulp
 from pulpcore.app.replica import ReplicaContext, distros_lock_uri
 from pulpcore.constants import TASK_STATES
-from pulpcore.exceptions import ExternalServiceError
+from pulpcore.exceptions import ExternalServiceError, ReplicateError
 from pulpcore.tasking.tasks import dispatch
 
 
@@ -128,7 +129,7 @@ def replicate_distributions(server_pk, q_select=None, **kwargs):
             # a full (non-overridden) replication runs.
             if q_select is None:
                 replicator.remove_missing(distro_names)
-    except GluePulpException as e:
+    except (GluePulpException, requests.exceptions.RequestException) as e:
         raise ExternalServiceError(service_name=server.base_url, details=str(e))
 
     dispatch(
@@ -149,8 +150,8 @@ def finalize_replication(server_pk, distro_repo_pairs, **kwargs):
         for t in failed_tasks:
             error_desc = t.error.get("description", "unknown error") if t.error else t.state
             details.append(f"  {t.name}: {error_desc}")
-        raise Exception(
-            "Replication failed. {} subtask(s) did not complete successfully:\n{}".format(
+        raise ReplicateError(
+            details="{} subtask(s) did not complete successfully:\n{}".format(
                 failed_tasks.count(), "\n".join(details)
             )
         )

pulpcore/app/tasks/test.py

@@ -7,9 +7,23 @@
 
 from pulpcore.app.models import Task, TaskGroup
 from pulpcore.constants import TASK_STATES
+from pulpcore.exceptions import PulpException
 from pulpcore.tasking.tasks import dispatch
 
 
+class PulpTestError(PulpException):
+    """A task error used by the test tasks below."""
+
+    error_code = "PLP0000"
+
+    def __init__(self, message):
+        super().__init__()
+        self.message = message
+
+    def __str__(self):
+        return self.message
+
+
 def dummy_task():
     """Dummy task, that can be used in tests."""
     pass
@@ -84,23 +98,23 @@ def missing_worker():
 
 def failing_task(error_message="Task intentionally failed"):
     """
-    A task that always raises a RuntimeError.
+    A task that always raises a PulpTestError.
 
     This task is used for testing error handling in worker task execution.
 
     Args:
-        error_message (str): The error message to include in the RuntimeError
+        error_message (str): The error message to include in the exception
     """
-    raise RuntimeError(error_message)
+    raise PulpTestError(error_message)
 
 
 async def afailing_task(error_message="Task intentionally failed"):
     """
-    An async task that always raises a RuntimeError.
+    An async task that always raises a PulpTestError.
 
     This task is used for testing error handling in immediate task execution.
 
     Args:
-        error_message (str): The error message to include in the RuntimeError
+        error_message (str): The error message to include in the exception
     """
-    raise RuntimeError(error_message)
+    raise PulpTestError(error_message)

pulpcore/exceptions/__init__.py

@@ -24,6 +24,7 @@
     SslConnectionError,
     RemoteConnectionError,
     FeatureNotImplementedError,
+    ProtectedResourceError,
 )
 from .validation import (
     ContentOverwriteError,

pulpcore/exceptions/base.py

@@ -324,8 +324,19 @@ class ReplicateError(PulpException):
 
     error_code = "PLP0018"
 
+    def __init__(self, details=None):
+        """
+        :param details: Additional details about the failure
+        :type details: str or None
+        """
+        super().__init__()
+        self.details = details
+
     def __str__(self):
-        return f"[{self.error_code}] " + _("Replication failed")
+        msg = _("Replication failed")
+        if self.details:
+            msg += f": {self.details}"
+        return f"[{self.error_code}] " + msg
 
 
 class TaskConfigurationError(PulpException):
@@ -445,3 +456,25 @@ def __init__(self, message):
 
     def __str__(self):
         return f"[{self.error_code}] {self.message}"
+
+
+class ProtectedResourceError(PulpException):
+    """
+    Raised when an object cannot be deleted because other objects still reference it.
+    """
+
+    error_code = "PLP0029"
+
+    def __init__(self, details=None):
+        """
+        :param details: Additional details about the protecting references
+        :type details: str or None
+        """
+        super().__init__()
+        self.details = details
+
+    def __str__(self):
+        msg = _("Cannot delete the object because it is still referenced by other objects")
+        if self.details:
+            msg += f": {self.details}"
+        return f"[{self.error_code}] " + msg

pulpcore/plugin/exceptions.py

@@ -6,6 +6,7 @@
     HttpResponseError,
     InvalidSignatureError,
     MissingDigestValidationError,
+    ProtectedResourceError,
     PublishError,
     PulpException,
     RemoteConnectionError,
@@ -38,4 +39,5 @@
     "HttpResponseError",
     "SslConnectionError",
     "RemoteConnectionError",
+    "ProtectedResourceError",
 ]

pulpcore/tests/functional/api/test_task_purge.py

@@ -28,7 +28,7 @@ def good_and_bad_task(
     good_task = monitor_task(dispatch_task("pulpcore.app.tasks.test.sleep", args=(0,)))
     assert good_task.state == "completed"
 
-    bad_task_href = dispatch_task("pulpcore.app.tasks.test.sleep", args=(-1,))
+    bad_task_href = dispatch_task("pulpcore.app.tasks.test.failing_task")
     with pytest.raises(PulpTaskError):
         monitor_task(bad_task_href)
     bad_task = pulpcore_bindings.TasksApi.read(bad_task_href)

pulpcore/tests/functional/api/test_tasking.py

@@ -559,7 +559,7 @@ def test_timeouts_on_api_worker(self, pulpcore_bindings, dispatch_task):
 @pytest.mark.parallel
 def test_failing_immediate_task_error_handling(dispatch_task, monitor_task):
     """
-    GIVEN a task that raises a RuntimeError
+    GIVEN a task that raises a PulpException
     AND the task is an async function
     WHEN dispatching the task as immediate and deferred
     THEN the task fails with the correct error message
@@ -585,7 +585,7 @@ def test_failing_immediate_task_error_handling(dispatch_task, monitor_task):
 @pytest.mark.parallel
 def test_failing_worker_task_error_handling(dispatch_task, monitor_task):
     """
-    GIVEN a task that raises a RuntimeError
+    GIVEN a task that raises a PulpException
     AND the task is a sync function
     WHEN dispatching the task as deferred (executes on worker)
     THEN the task fails with the correct error message

pulpcore/tests/unit/tasking/test_delete_tasks.py

@@ -0,0 +1,50 @@
+"""Unit tests for ProtectedError handling in the generic delete tasks."""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from asgiref.sync import async_to_sync
+from django.db.models.deletion import ProtectedError
+
+from pulpcore.app.tasks import base
+from pulpcore.exceptions import ProtectedResourceError
+
+
+def _patch_plugin_config(monkeypatch, instance):
+    """Make the delete tasks resolve to a model whose .get() returns `instance`."""
+    serializer_class = MagicMock()
+    serializer_class.Meta.model.objects.get.return_value = instance
+    serializer_class.Meta.model.objects.aget = AsyncMock(return_value=instance)
+    plugin_config = MagicMock()
+    plugin_config.named_serializers = {"FakeSerializer": serializer_class}
+    monkeypatch.setattr(base, "get_plugin_config", lambda app_label: plugin_config)
+
+
+@pytest.mark.django_db
+def test_general_delete_wraps_protected_error(monkeypatch):
+    instance = MagicMock()
+    instance.delete.side_effect = ProtectedError("still referenced", set())
+    _patch_plugin_config(monkeypatch, instance)
+
+    with pytest.raises(ProtectedResourceError):
+        base.general_delete("some-pk", "core", "FakeSerializer")
+
+
+@pytest.mark.django_db
+def test_general_multi_delete_wraps_protected_error(monkeypatch):
+    instance = MagicMock()
+    instance.delete.side_effect = ProtectedError("still referenced", set())
+    _patch_plugin_config(monkeypatch, instance)
+
+    with pytest.raises(ProtectedResourceError):
+        base.general_multi_delete([("some-pk", "core", "FakeSerializer")])
+
+
+@pytest.mark.django_db
+def test_ageneral_delete_wraps_protected_error(monkeypatch):
+    instance = MagicMock()
+    instance.adelete = AsyncMock(side_effect=ProtectedError("still referenced", set()))
+    _patch_plugin_config(monkeypatch, instance)
+
+    with pytest.raises(ProtectedResourceError):
+        async_to_sync(base.ageneral_delete)("some-pk", "core", "FakeSerializer")