[PR #7712/e7444266 backport][3.110] Added additional locking during replication

CHANGES/7614.bugfix

@@ -0,0 +1 @@
+Added a distributions lock to `finalize_replication` to prevent concurrent finalize tasks from different upstream servers from racing on the same distribution objects.

pulpcore/app/replica.py

@@ -17,6 +17,10 @@
 _logger = logging.getLogger(__name__)
 
 
+def distros_lock_uri(domain_id):
+    return f"pdrn:{domain_id}:distributions"
+
+
 class ReplicaContext(PulpContext):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
@@ -61,7 +65,7 @@ def __init__(self, pulp_ctx, task_group, tls_settings, server):
         self.tls_settings = tls_settings
         self.server = server
         self.domain = get_domain()
-        self.distros_uris = [f"pdrn:{self.domain.pulp_id}:distributions"]
+        self.distros_uris = [distros_lock_uri(self.domain.pulp_id)]
 
     @staticmethod
     def needs_update(fields_dict, model_instance):

pulpcore/app/tasks/replica.py

@@ -10,7 +10,7 @@
 
 from pulpcore.app.apps import PulpAppConfig, pulp_plugin_configs
 from pulpcore.app.models import Distribution, Repository, Task, TaskGroup, UpstreamPulp
-from pulpcore.app.replica import ReplicaContext
+from pulpcore.app.replica import ReplicaContext, distros_lock_uri
 from pulpcore.constants import TASK_STATES
 from pulpcore.tasking.tasks import dispatch
 
@@ -111,7 +111,7 @@ def replicate_distributions(server_pk):
     dispatch(
         finalize_replication,
         task_group=task_group,
-        exclusive_resources=[server],
+        exclusive_resources=[server, distros_lock_uri(server.pulp_domain_id)],
         args=[server.pk, distro_repo_pairs],
     )