Fix replication to inherit retain_repo_versions from primary

CHANGES/7921.bugfix

@@ -0,0 +1 @@
+Fixed replicas to inherit `retain_repo_versions` setting from the primary Pulp instance during replication.

pulp_file/app/replica.py

@@ -43,8 +43,10 @@ def url(self, upstream_distribution):
 
         return f"{upstream_distribution['base_url']}{manifest}"
 
-    def repository_extra_fields(self, remote):
-        return dict(manifest=remote.url.split("/")[-1], autopublish=False)
+    def repository_extra_fields(self, remote, **kwargs):
+        fields = super().repository_extra_fields(remote, **kwargs)
+        fields.update(dict(manifest=remote.url.split("/")[-1], autopublish=False))
+        return fields
 
     def sync_params(self, repository, remote):
         return dict(

pulpcore/app/replica.py

@@ -148,11 +148,61 @@ def create_or_update_remote(self, upstream_distribution):
 
         return remote
 
-    def repository_extra_fields(self, remote):
-        return {}
+    def _extract_repository_href(self, upstream_distribution):
+        """
+        Extract repository href from upstream distribution.
+
+        The distribution may reference a repository directly, or via repository_version,
+        or may only have a publication (in which case we can't get the repo).
 
-    def create_or_update_repository(self, remote):
-        repo_fields_dict = self.repository_extra_fields(remote)
+        Args:
+            upstream_distribution: Dict containing distribution data from upstream
+
+        Returns:
+            str or None: Repository href if available
+        """
+        if upstream_distribution.get("repository"):
+            return upstream_distribution["repository"]
+        elif upstream_distribution.get("repository_version"):
+            # Extract repo href from repo version href
+            # Format: /pulp/api/v3/repositories/rpm/rpm/{uuid}/versions/{num}/
+            # We want: /pulp/api/v3/repositories/rpm/rpm/{uuid}/
+            return upstream_distribution["repository_version"].rsplit("versions/", 1)[0]
+        return None
+
+    def repository_extra_fields(self, remote, upstream_distribution=None):
+        """
+        Return extra fields to set on the repository during replication.
+
+        Args:
+            remote: The remote object being used for replication
+            upstream_distribution: Optional upstream distribution dict containing repository href
+
+        Returns:
+            dict: Fields to set on the repository
+        """
+        fields = {}
+
+        # Fetch retain_repo_versions from upstream repository if available
+        if upstream_distribution:
+            repo_href = self._extract_repository_href(upstream_distribution)
+            if repo_href:
+                try:
+                    upstream_repo = self.repository_ctx_cls(self.pulp_ctx, repo_href).entity
+                    if "retain_repo_versions" in upstream_repo:
+                        # Copy the value from upstream (None means unlimited, which is valid)
+                        fields["retain_repo_versions"] = upstream_repo["retain_repo_versions"]
+                except Exception as e:
+                    # Log but don't fail - retain_repo_versions is not critical
+                    _logger.warning(
+                        f"Failed to fetch retain_repo_versions from upstream repository "
+                        f"{repo_href}: {e}"
+                    )
+
+        return fields
+
+    def create_or_update_repository(self, remote, upstream_distribution=None):
+        repo_fields_dict = self.repository_extra_fields(remote, upstream_distribution)
         repo_fields_dict["pulp_labels"] = self.labels(remote)
         try:
             repository = self.repository_model_cls.objects.get(

pulpcore/app/tasks/replica.py

@@ -99,7 +99,9 @@ def replicate_distributions(server_pk, q_select=None, **kwargs):
                     # let it fall through the cracks and be cleaned up below.
                     continue
                 # Check if there is already a repository
-                repository = replicator.create_or_update_repository(remote=remote)
+                repository = replicator.create_or_update_repository(
+                    remote=remote, upstream_distribution=distro
+                )
                 if not repository:
                     # No update occurred because server.policy==LABELED and there was
                     # an already existing local repository with the same name

pulpcore/tests/functional/api/test_replication.py

@@ -1407,3 +1407,118 @@ def test_replicate_policy(
                 assert "UpstreamPulp" not in distro.pulp_labels
     else:
         assert result.count == 0
+
+
+@pytest.mark.parallel
+def test_replication_retain_repo_versions(
+    domain_factory,
+    bindings_cfg,
+    pulpcore_bindings,
+    file_bindings,
+    monitor_task,
+    monitor_task_group,
+    pulp_settings,
+    gen_object_with_cleanup,
+    file_distribution_factory,
+    file_repository_factory,
+    tmp_path,
+    add_domain_objects_to_cleanup,
+):
+    """Test that retain_repo_versions setting is replicated from upstream."""
+
+    # Create source domain with a repository that has retain_repo_versions=5
+    source_domain = domain_factory()
+    add_domain_objects_to_cleanup(source_domain)
+
+    # Create repository with retain_repo_versions set
+    repository = file_repository_factory(pulp_domain=source_domain.name, retain_repo_versions=5)
+
+    # Create some content and add to repository
+    file_path = tmp_path / "file1.txt"
+    file_path.write_text("content1")
+    monitor_task(
+        file_bindings.ContentFilesApi.create(
+            file=str(file_path),
+            relative_path="file1.txt",
+            repository=repository.pulp_href,
+            pulp_domain=source_domain.name,
+        ).task
+    )
+
+    # Create distribution
+    file_distribution_factory(pulp_domain=source_domain.name, repository=repository.pulp_href)
+
+    # Create replica domain
+    replica_domain = domain_factory()
+    add_domain_objects_to_cleanup(replica_domain)
+
+    # Create UpstreamPulp and replicate
+    upstream_pulp_body = {
+        "name": str(uuid.uuid4()),
+        "base_url": bindings_cfg.host,
+        "api_root": pulp_settings.API_ROOT,
+        "domain": source_domain.name,
+        "username": bindings_cfg.username,
+        "password": bindings_cfg.password,
+    }
+    upstream_pulp = gen_object_with_cleanup(
+        pulpcore_bindings.UpstreamPulpsApi,
+        upstream_pulp_body,
+        pulp_domain=replica_domain.name,
+    )
+
+    # Run replication
+    response = pulpcore_bindings.UpstreamPulpsApi.replicate(
+        upstream_pulp.pulp_href, pulpcore_bindings.module.UpstreamPulpReplicate()
+    )
+    task_group = monitor_task_group(response.task_group)
+
+    # Verify replication succeeded
+    for task in task_group.tasks:
+        assert task.state == "completed"
+
+    # Verify retain_repo_versions was replicated
+    replica_repos = file_bindings.RepositoriesFileApi.list(
+        pulp_domain=replica_domain.name, name=repository.name
+    )
+    assert replica_repos.count == 1
+    replica_repo = replica_repos.results[0]
+    assert replica_repo.retain_repo_versions == 5, (
+        f"Expected retain_repo_versions=5, got {replica_repo.retain_repo_versions}"
+    )
+
+    # Test with None value (unlimited retention)
+    repository2 = file_repository_factory(
+        pulp_domain=source_domain.name, name="test-repo-unlimited", retain_repo_versions=None
+    )
+    file_path2 = tmp_path / "file2.txt"
+    file_path2.write_text("content2")
+    monitor_task(
+        file_bindings.ContentFilesApi.create(
+            file=str(file_path2),
+            relative_path="file2.txt",
+            repository=repository2.pulp_href,
+            pulp_domain=source_domain.name,
+        ).task
+    )
+    file_distribution_factory(
+        pulp_domain=source_domain.name, repository=repository2.pulp_href, name="test-distro-2"
+    )
+
+    # Replicate again
+    response = pulpcore_bindings.UpstreamPulpsApi.replicate(
+        upstream_pulp.pulp_href, pulpcore_bindings.module.UpstreamPulpReplicate()
+    )
+    task_group = monitor_task_group(response.task_group)
+    for task in task_group.tasks:
+        assert task.state == "completed"
+
+    # Verify None value was replicated
+    replica_repos2 = file_bindings.RepositoriesFileApi.list(
+        pulp_domain=replica_domain.name, name=repository2.name
+    )
+    assert replica_repos2.count == 1
+    replica_repo2 = replica_repos2.results[0]
+    assert replica_repo2.retain_repo_versions is None, (
+        f"Expected retain_repo_versions=None, got {replica_repo2.retain_repo_versions}"
+    )