task: add clean up task for parent pids that are not deleted

invenio_rdm_records/services/components/pids.py

@@ -17,7 +17,7 @@
 from invenio_drafts_resources.services.records.uow import ParentRecordCommitOp
 from invenio_records_resources.services.uow import TaskOp
 
-from ..pids.tasks import register_or_update_pid
+from ..pids.tasks import cleanup_parent_pids, register_or_update_pid
 
 
 class PIDsComponent(ServiceComponent):
@@ -212,6 +212,11 @@ def delete_record(self, identity, data=None, record=None, uow=None):
             self.service.pids.parent_pid_manager.discard_all(
                 parent_pids, soft_delete=True
             )
+        else:
+            # We're sending a task in case there is a race condition with two
+            # versions being deleted at the same time to ensure that we have
+            # consistent database state
+            self.uow.register(TaskOp(cleanup_parent_pids, record["id"]))
 
         # Async register/update tasks after transaction commit.
         for scheme in parent_pids.keys():

invenio_rdm_records/services/pids/tasks.py

@@ -7,6 +7,8 @@
 
 """RDM PIDs Service tasks."""
 
+from copy import deepcopy
+
 from celery import shared_task
 from invenio_access.permissions import system_identity
 
@@ -22,3 +24,16 @@ def register_or_update_pid(recid, scheme, parent=False):
         scheme=scheme,
         parent=parent,
     )
+
+
+@shared_task(ignore_result=True)
+def cleanup_parent_pids(recid):
+    """Clean up parent PIDs."""
+    record_cls = current_rdm_records.records_service
+    record = record_cls.pid.resolve(recid, with_deleted=True)
+    if record.deleted:
+        parent_pids = deepcopy(record.parent.get("pids", {}))
+        if record_cls.next_latest_published_record_by_parent(record.parent) is None:
+            record_cls.pids.parent_pid_manager.discard_all(
+                parent_pids, soft_delete=True
+            )