new approach

Author: dralley

pulpcore/app/replica.py

@@ -174,12 +174,10 @@ def distribution_extra_fields(self, repository, upstream_distribution):
         """
         Return the fields that need to be updated/cleared on distributions for idempotence.
 
-        Note: repository_version is NOT included here. It is set atomically in
-        finalize_replication after all syncs complete.
+        Note: repository, publication, and repository_version are NOT included here.
+        They are all updated atomically in finalize_replication after all syncs complete.
         """
         return {
-            "repository": None,
-            "publication": None,
             "base_path": upstream_distribution["base_path"],
         }
 
@@ -192,8 +190,6 @@ def create_or_update_distribution(self, repository, upstream_distribution):
             )
             if not self._is_managed(distro):
                 return None
-            # Clear repository and publication so they don't conflict when
-            # finalize_replication sets repository_version atomically.
             needs_update = self.needs_update(distribution_data, distro)
             if needs_update:
                 dispatch(
@@ -208,13 +204,8 @@ def create_or_update_distribution(self, repository, upstream_distribution):
                     },
                 )
         except self.distribution_model_cls.DoesNotExist:
-            # Don't set repository or publication for new distributions —
-            # repository_version will be set in finalize_replication after sync completes.
-            create_data = {
-                k: v for k, v in distribution_data.items() if k not in ("repository", "publication")
-            }
+            create_data = dict(distribution_data)
             create_data["name"] = upstream_distribution["name"]
-            create_data["pulp_labels"] = distribution_data["pulp_labels"]
             dispatch(
                 general_create,
                 task_group=self.task_group,

pulpcore/app/tasks/replica.py

@@ -124,16 +124,24 @@ def finalize_replication(server_pk, distro_repo_pairs):
     if task_group.tasks.exclude(pk=task.pk).exclude(state=TASK_STATES.COMPLETED).exists():
         raise Exception("Replication failed.")
 
-    # Atomically update all managed distributions to point to their repo's latest version.
+    # Atomically update all managed distributions to point to their repo's latest version,
+    # clearing any previous repository or publication references.
     with transaction.atomic():
         for distro_name, repo_pk in distro_repo_pairs:
             distro = Distribution.objects.get(name=distro_name, pulp_domain=server.pulp_domain)
             repo = Repository.objects.get(pk=repo_pk)
             latest_version = repo.latest_version()
             if latest_version:
-                if distro.repository_version != latest_version:
+                needs_update = (
+                    distro.repository_version != latest_version
+                    or distro.repository is not None
+                    or distro.publication is not None
+                )
+                if needs_update:
+                    distro.repository = None
+                    distro.publication = None
                     distro.repository_version = latest_version
-                    distro.save(update_fields=["repository_version"])
+                    distro.save(update_fields=["repository", "publication", "repository_version"])
 
     # Record timestamp of last successful replication.
     started_at = task_group.tasks.aggregate(Min("started_at"))["started_at__min"]