Protect distributed repo versions from deletion

This change protects repo versions from being deleted if they are being
used to distribute content. It also protects repo versions from being
cleaned up by the retain repo versions code.

fixes pulp#2705

CHANGES/2705.bugfix

@@ -0,0 +1,2 @@
+Repo versions are now protected from deletion if they are being used by Pulp to distribute content.
+Users must first update any necessary distributions before deleting a protected repo version.

docs/workflows/repo-versioning.rst

@@ -16,6 +16,9 @@ greater than or equal to 1.
 Setting retain_repo_versions to 1 effectively disables repository versioning since Pulp will only
 store the latest version.
 
+Cleanup will ignore any repo versions that are being served directly via a distribution or via a
+publication.
+
 To update this field for a file Repository called myrepo, simply call:
 
 ```

pulpcore/app/models/repository.py

@@ -26,7 +26,7 @@
     get_domain_pk,
     cache_key,
 )
-from pulpcore.constants import ALL_KNOWN_CONTENT_CHECKSUMS
+from pulpcore.constants import ALL_KNOWN_CONTENT_CHECKSUMS, PROTECTED_REPO_VERSION_MESSAGE
 from pulpcore.download.factory import DownloaderFactory
 from pulpcore.exceptions import ResourceImmutableError
 
@@ -309,6 +309,42 @@ def artifacts_for_version(version):
         """
         return Artifact.objects.filter(content__pk__in=version.content)
 
+    def protected_versions(self):
+        """
+        Return repository versions that are protected.
+
+        A protected version is one that is being served by a distro directly or via publication.
+
+        Returns:
+            django.db.models.QuerySet: Repo versions which are protected.
+        """
+        from .publication import Distribution, Publication
+
+        # find all repo versions set on a distribution
+        qs = self.versions.filter(pk__in=Distribution.objects.values_list("repository_version_id"))
+
+        # find all repo versions with publications set on a distribution
+        qs |= self.versions.filter(
+            publication__pk__in=Distribution.objects.values_list("publication_id")
+        )
+
+        if distro := Distribution.objects.filter(repository=self.pk).first():
+            if distro.detail_model().SERVE_FROM_PUBLICATION:
+                # if the distro serves publications, protect the latest published repo version
+                version = self.versions.filter(
+                    pk__in=Publication.objects.filter(complete=True).values_list(
+                        "repository_version_id"
+                    )
+                ).last()
+            else:
+                # if the distro does not serve publications, use the latest repo version
+                version = self.latest_version()
+
+            if version:
+                qs |= self.versions.filter(pk=version.pk)
+
+        return qs.distinct()
+
     @hook(AFTER_UPDATE, when="retain_repo_versions", has_changed=True)
     def _cleanup_old_versions_hook(self):
         # Do not attempt to clean up anything, while there is a transaction involving repo versions
@@ -325,10 +361,9 @@ def cleanup_old_versions(self):
                 _("Attempt to cleanup old versions, while a new version is in flight.")
             )
         if self.retain_repo_versions:
-            # Consider only completed versions for cleanup
-            for version in self.versions.complete().order_by("-number")[
-                self.retain_repo_versions :
-            ]:
+            # Consider only completed versions that aren't protected for cleanup
+            versions = self.versions.complete().exclude(pk__in=self.protected_versions())
+            for version in versions.order_by("-number")[self.retain_repo_versions :]:
                 _logger.info(
                     "Deleting repository version {} due to version retention limit.".format(version)
                 )
@@ -1062,6 +1097,12 @@ def _squash(self, repo_relations, next_version):
         # Update next version's counts as they have been modified
         next_version._compute_counts()
 
+    @hook(BEFORE_DELETE)
+    def check_protected(self):
+        """Check if a repo version is protected before trying to delete it."""
+        if self in self.repository.protected_versions():
+            raise Exception(PROTECTED_REPO_VERSION_MESSAGE)
+
     def delete(self, **kwargs):
         """
         Deletes a RepositoryVersion

pulpcore/app/viewsets/repository.py

@@ -10,6 +10,7 @@
 from rest_framework.viewsets import GenericViewSet
 from urllib.parse import urlparse
 
+from pulpcore.constants import PROTECTED_REPO_VERSION_MESSAGE
 from pulpcore.filters import BaseFilterSet
 from pulpcore.app import tasks
 from pulpcore.app.models import (
@@ -296,6 +297,9 @@ def destroy(self, request, repository_pk, number):
         """
         version = self.get_object()
 
+        if version in version.repository.protected_versions():
+            raise serializers.ValidationError(PROTECTED_REPO_VERSION_MESSAGE)
+
         task = dispatch(
             tasks.repository.delete_version,
             exclusive_resources=[version.repository],

pulpcore/constants.py

@@ -1,3 +1,4 @@
+from gettext import gettext as _
 from pathlib import Path
 from types import SimpleNamespace
 
@@ -104,3 +105,9 @@
     "storages.backends.azure_storage.AzureStorage": AZURE_RESPONSE_HEADER_MAP,
     "storages.backends.gcloud.GoogleCloudStorage": GCS_RESPONSE_HEADER_MAP,
 }
+
+# Message users receive when attempting to delete a protected repo version
+PROTECTED_REPO_VERSION_MESSAGE = _(
+    "The repository version cannot be deleted because it (or its publications) are currently being "
+    "used to distribute content. Please update the necessary distributions first."
+)

pulpcore/tests/functional/api/using_plugin/test_repo_versions.py

@@ -34,6 +34,28 @@ def file_9_contents(
     return content_units
 
 
+@pytest.fixture
+def file_repository_content(
+    file_remote_ssl_factory,
+    file_repository_factory,
+    file_repository_api_client,
+    file_content_api_client,
+    basic_manifest_path,
+    monitor_task,
+):
+    """Create some content that was synced into a repo on-demand."""
+    remote = file_remote_ssl_factory(manifest_path=basic_manifest_path, policy="on_demand")
+    base_repo = file_repository_factory()
+    task = file_repository_api_client.sync(base_repo.pulp_href, {"remote": remote.pulp_href}).task
+    monitor_task(task)
+    base_repo = file_repository_api_client.read(base_repo.pulp_href)
+    assert base_repo.latest_version_href[-2] == "1"
+    contents = file_content_api_client.list(repository_version=base_repo.latest_version_href)
+    assert contents.count == 3
+
+    return contents
+
+
 @pytest.mark.parallel
 def test_add_remove_content(
     file_repository_api_client,
@@ -626,23 +648,17 @@ def test_filter_artifacts(
 
 
 @pytest.mark.parallel
-def test_delete_repo_version_resources(
+def test_delete_repo_version_publication(
     file_repository_api_client,
     file_repository_version_api_client,
     file_repository_factory,
     file_remote_ssl_factory,
-    file_distribution_factory,
     basic_manifest_path,
     file_publication_api_client,
-    file_distribution_api_client,
     gen_object_with_cleanup,
     monitor_task,
 ):
-    """Test whether removing a repository version affects related resources.
-
-    Test whether removing a repository version will remove a related Publication.
-    Test whether removing a repository version a Distribution will not be removed.
-    """
+    """Test that removing a repo version will delete its publication."""
     file_repo = file_repository_factory()
     remote = file_remote_ssl_factory(manifest_path=basic_manifest_path, policy="on_demand")
     task = file_repository_api_client.sync(file_repo.pulp_href, {"remote": remote.pulp_href}).task
@@ -654,9 +670,6 @@ def test_delete_repo_version_resources(
     publication = gen_object_with_cleanup(file_publication_api_client, pub_body)
     assert publication.repository_version == repo.latest_version_href
 
-    distribution = file_distribution_factory(publication=publication.pulp_href)
-    assert distribution.publication == publication.pulp_href
-
     # delete repo version used to create publication
     file_repository_version_api_client.delete(repo.latest_version_href)
 
@@ -665,8 +678,53 @@ def test_delete_repo_version_resources(
 
     assert e.value.status == 404
 
-    updated_distribution = file_distribution_api_client.read(distribution.pulp_href)
-    assert updated_distribution.publication is None
+
+@pytest.mark.parallel
+def test_delete_protected_repo_version(
+    file_repository_api_client,
+    file_repository_version_api_client,
+    file_repository_factory,
+    file_remote_ssl_factory,
+    file_distribution_factory,
+    basic_manifest_path,
+    file_publication_api_client,
+    file_distribution_api_client,
+    gen_object_with_cleanup,
+    monitor_task,
+):
+    """Test that removing a repo version fails if its publication is distributed."""
+    file_repo = file_repository_factory()
+    remote = file_remote_ssl_factory(manifest_path=basic_manifest_path, policy="on_demand")
+    task = file_repository_api_client.sync(file_repo.pulp_href, {"remote": remote.pulp_href}).task
+    monitor_task(task)
+    repo = file_repository_api_client.read(file_repo.pulp_href)
+    assert repo.latest_version_href[-2] == "1"
+
+    pub_body = {"repository": repo.pulp_href}
+    publication = gen_object_with_cleanup(file_publication_api_client, pub_body)
+    assert publication.repository_version == repo.latest_version_href
+
+    distribution = file_distribution_factory(publication=publication.pulp_href)
+    assert distribution.publication == publication.pulp_href
+
+    # deleting a protected repo version fails
+    with pytest.raises(ApiException) as e:
+        file_repository_version_api_client.delete(repo.latest_version_href)
+    assert e.value.status == 400
+    assert "The repository version cannot be deleted" in e.value.body
+
+    # unset the publication for the distribution
+    task = file_distribution_api_client.partial_update(
+        distribution.pulp_href, {"publication": ""}
+    ).task
+    monitor_task(task)
+
+    # and then delete the repo version
+    task = file_repository_version_api_client.delete(repo.latest_version_href).task
+    monitor_task(task)
+    with pytest.raises(ApiException) as e:
+        file_repository_version_api_client.read(repo.latest_version_href)
+    assert e.value.status == 404
 
 
 @pytest.mark.parallel
@@ -735,6 +793,7 @@ def test_clear_all_units_repo_version(
 def test_repo_version_retention(
     file_repository_api_client,
     file_repository_version_api_client,
+    file_repository_content,
     file_content_api_client,
     file_publication_api_client,
     file_repository_factory,
@@ -745,14 +804,7 @@ def test_repo_version_retention(
 ):
     """Test retain_repo_versions for repositories."""
     # Setup
-    remote = file_remote_ssl_factory(manifest_path=basic_manifest_path, policy="on_demand")
-    base_repo = file_repository_factory()
-    task = file_repository_api_client.sync(base_repo.pulp_href, {"remote": remote.pulp_href}).task
-    monitor_task(task)
-    base_repo = file_repository_api_client.read(base_repo.pulp_href)
-    assert base_repo.latest_version_href[-2] == "1"
-    contents = file_content_api_client.list(repository_version=base_repo.latest_version_href)
-    assert contents.count == 3
+    contents = file_repository_content
 
     # Test repo version retention.
     repo = file_repository_factory(retain_repo_versions=1)
@@ -825,6 +877,60 @@ def test_repo_version_retention(
     assert len(manifest_files) == contents.count
 
 
+@pytest.mark.parallel
+def test_repo_versions_protected_from_cleanup(
+    file_repository_api_client,
+    file_repository_version_api_client,
+    file_repository_content,
+    file_publication_api_client,
+    file_repository_factory,
+    file_distribution_factory,
+    gen_object_with_cleanup,
+    monitor_task,
+):
+    """Test that distributed repo versions are protected from retain_repo_versions."""
+
+    def _modify_and_validate(repo, content, expected_version, expected_total):
+        task = file_repository_api_client.modify(
+            repo.pulp_href, {"add_content_units": [content.pulp_href]}
+        ).task
+        monitor_task(task)
+
+        repo = file_repository_api_client.read(repo.pulp_href)
+        assert repo.latest_version_href[-2] == expected_version
+
+        versions = file_repository_version_api_client.list(repo.pulp_href)
+        assert versions.count == expected_total
+
+        return repo
+
+    # Setup
+    contents = file_repository_content
+    repo = file_repository_factory(retain_repo_versions=1)
+
+    # Publish and distribute version 0
+    publication = gen_object_with_cleanup(
+        file_publication_api_client, {"repository_version": repo.latest_version_href}
+    )
+    file_distribution_factory(publication=publication.pulp_href)
+
+    # Version 0 is protected since it's distributed
+    repo = _modify_and_validate(repo, contents.results[0], "1", 2)
+
+    # Create a new publication and distribution which protects version 1 from deletion
+    file_distribution_factory(repository=repo.pulp_href)
+    publication = gen_object_with_cleanup(
+        file_publication_api_client, {"repository_version": repo.latest_version_href}
+    )
+    file_distribution_factory(publication=publication.pulp_href)
+
+    # Create version 2 and there should be 3 versions now (2 protected)
+    repo = _modify_and_validate(repo, contents.results[1], "2", 3)
+
+    # Version 2 will be removed since we're creating version 3 and it's not protected
+    _modify_and_validate(repo, contents.results[2], "3", 3)
+
+
 @pytest.mark.parallel
 def test_content_in_repository_version_view(
     file_repository_api_client,