Reworked the task purge feature

When an object could not be deleted, the purge task would fail.
This has been reworked  so that the purge continues to delete
other tasks and informs the user about tasks that could not be
deleted.

fixes: pulp#3741

CHANGES/3741.feature

@@ -0,0 +1 @@
+Task purge now continues when encountering a task that could not be deleted. The user is informed about any such tasks.

pulpcore/app/tasks/purge.py

@@ -1,6 +1,8 @@
 from gettext import gettext as _
+from logging import getLogger
 
 from django_currentuser.middleware import get_current_authenticated_user
+from django.db.models.deletion import ProtectedError
 from pulpcore.app.models import (
     ProgressReport,
     Task,
@@ -9,6 +11,8 @@
 from pulpcore.app.util import get_domain
 from pulpcore.constants import TASK_STATES
 
+log = getLogger(__name__)
+
 # Delete 1K at a time - better to use less memory, and take a little longer, with a utility
 # function like this.
 DELETE_LIMIT = 1000
@@ -101,15 +105,39 @@ def purge(finished_before, states):
     pb.save()
     details_reports[TASK_KEY] = pb
 
+    # Build and save a progress-report for objects that couldn't be deleted
+    error_pb = ProgressReport(
+        message=_("Failed to purge task-objects"),
+        total=None,
+        code="purge.tasks.error",
+        done=0,
+    )
+    error_pb.save()
+    # Also keep a list of PKs of objects we've already deleted, so we don't try to delete them twice
+    pks_deleted = []
+
     # Our delete-query is going to deal with "the first DELETE_LIMIT tasks that match our
     # criteria", looping until we've deleted everything that fits our parameters
     units_deleted = 1
     # Until our query returns "No tasks deleted", add results into totals and Do It Again
     while units_deleted > 0:
-        units_deleted, details = Task.objects.filter(
-            pk__in=candidate_qs[:DELETE_LIMIT].values_list("pk", flat=True)
-        ).delete()
-        _details_reporting(details_reports, details, totals_pb)
+        # Get a list of candidate objects to delete and reset the counter
+        candidate_pks = candidate_qs.exclude(pk__in=pks_deleted).values_list("pk", flat=True)
+        units_deleted = 0
+
+        # Loop though the candidate objects and delete them one-by-one
+        for pk in candidate_pks[:DELETE_LIMIT]:
+            try:
+                obj = Task.objects.get(pk=pk)
+                count, details = obj.delete()
+                units_deleted += count
+                _details_reporting(details_reports, details, totals_pb)
+            except ProtectedError as e:
+                # Object could not be deleted due to foreign key constraint.
+                # Log the details of the object.
+                error_pb.done += 1
+                pks_deleted.append(pk)
+                log.info(e)
 
     # Complete the progress-reports for the specific entities deleted
     for key, pb in details_reports.items():
@@ -121,3 +149,8 @@ def purge(finished_before, states):
     totals_pb.total = totals_pb.done
     totals_pb.state = TASK_STATES.COMPLETED
     totals_pb.save()
+
+    # Complete the error-ProgressReport
+    error_pb.total = error_pb.done
+    error_pb.state = TASK_STATES.COMPLETED
+    error_pb.save()