Fixed data-fixup-during-sync deadlock.

fixes pulp#2980 [nocoverage]
ggainey · Mar 10, 2023 · 5d7780e · 5d7780e
1 parent 7baced8
commit 5d7780e
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 11 deletions.
diff --git a/CHANGES/2980.bugfix b/CHANGES/2980.bugfix
@@ -0,0 +1 @@
+Fixed a deadlock during concurrent syncs of rpm-repos that need data fixups.
diff --git a/pulp_rpm/app/tasks/synchronizing.py b/pulp_rpm/app/tasks/synchronizing.py
@@ -1576,15 +1576,8 @@ async def run(self):
                     content_q_by_type[model_type] = content_q_by_type[model_type] | unit_q
                     d_content_by_nat_key[d_content.content.natural_key()].append(d_content)
 
+            modified_rslts_by_pid = {}  # hold the to-be-saved results by-pid
             for model_type, content_q in content_q_by_type.items():
-                try:
-                    await sync_to_async(model_type.objects.filter(content_q).touch)()
-                except AttributeError:
-                    raise TypeError(
-                        "Plugins which declare custom ORM managers on their content classes "
-                        "should have those managers inherit from "
-                        "pulpcore.plugin.models.ContentManager."
-                    )
                 async for result in sync_to_async_iterable(
                     model_type.objects.filter(content_q).iterator()
                 ):
@@ -1594,13 +1587,13 @@ async def run(self):
                             # Fix saved snippet if malformed in DB, covers #2735
                             if result.snippet != d_content.content.snippet:
                                 result.snippet = d_content.content.snippet
-                                await sync_to_async(result.save)()
+                                modified_rslts_by_pid[result.pulp_id] = result
 
                         if model_type == ModulemdDefaults:
                             # Fix saved snippet if malformed in DB, covers #2735
                             if result.snippet != d_content.content.snippet:
                                 result.snippet = d_content.content.snippet
-                                await sync_to_async(result.save)()
+                                modified_rslts_by_pid[result.pulp_id] = result
 
                         if model_type == Package:
                             # changelogs coming out of the database are list[list],
@@ -1624,9 +1617,33 @@ async def run(self):
                                 d_content.content.files = result.files
                             if incorrect_changelogs or incorrect_modular_relation:
                                 log.debug("Updated data for package {}".format(result.nevra))
-                                await sync_to_async(result.save)()
+                                modified_rslts_by_pid[result.pulp_id] = result
                         # ==================================================================
                         d_content.content = result
 
+            # Save results in guaranteed-pid-order
+            modified_rslts_pids = sorted(modified_rslts_by_pid.keys())
+            for pid in modified_rslts_pids:
+                await sync_to_async(modified_rslts_by_pid[pid].save)()
+
+            # touch all affected content, **excluding** anything we just saved
+            query_types = sorted(content_q_by_type.keys(), key=lambda t: t.__name__)
+            for query_type in query_types:
+                try:
+                    # touch() in order to mark "last seen time" for the batch
+                    # NOTE: DO NOT include anything we fixed and saved above - that way
+                    # lie deadlocks!
+                    await sync_to_async(
+                        query_type.objects.filter(content_q_by_type[query_type])
+                        .exclude(pulp_id__in=modified_rslts_pids)
+                        .touch
+                    )()
+                except AttributeError:
+                    raise TypeError(
+                        "Plugins which declare custom ORM managers on their content classes "
+                        "should have those managers inherit from "
+                        "pulpcore.plugin.models.ContentManager."
+                    )
+
             for d_content in batch:
                 await self.put(d_content)