Fix a cache invalidation issue with Distributions serving publications indirectly via repository_version

CHANGES/7993.bugfix

@@ -0,0 +1 @@
+Fixed cache invalidation and DistributedPublication tracking for distributions serving via `repository_version` when publications are created or deleted.

pulpcore/app/models/publication.py

@@ -177,6 +177,20 @@ def delete(self, **kwargs):
             except Publication.DoesNotExist:
                 pass
 
+            # A distribution serving this publication's version directly (repository_version)
+            # resolves to the latest publication of that version. Invalidate those distributions
+            # when this is that latest publication.
+            try:
+                version_latest = Publication.objects.filter(
+                    repository_version=self.repository_version, complete=True
+                ).latest("pulp_created")
+                if self.pk == version_latest.pk:
+                    base_paths |= Distribution.objects.filter(
+                        repository_version=self.repository_version
+                    ).values_list("base_path", flat=True)
+            except Publication.DoesNotExist:
+                pass
+
             # Invalidate cache for all distributions serving this publication
             if base_paths:
                 Cache().delete(base_key=cache_key(base_paths))
@@ -234,17 +248,16 @@ def __exit__(self, exc_type, exc_val, exc_tb):
                 self.delete()
                 raise
 
-            # Create distributed publication for repository auto-publish scenario
+            # Refresh distributed publications for the auto-publish scenario. A distribution can
+            # distribute this publication indirectly either through its repository (latest
+            # publication of the latest version) or through its repository_version (latest
+            # publication of that version), without the distribution itself changing.
             if retain_distributed_pub_enabled():
-                for distro in Distribution.objects.filter(repository=self.repository):
-                    detail_distro = distro.cast()
-                    if not detail_distro.SERVE_FROM_PUBLICATION:
-                        continue
-                    _, _, latest_repo_publication = (
-                        detail_distro.get_repository_publication_and_version()
-                    )
-                    if self == latest_repo_publication:
-                        DistributedPublication(distribution=distro, publication=self).save()
+                for distro in Distribution.objects.filter(
+                    models.Q(repository=self.repository_version.repository)
+                    | models.Q(repository_version=self.repository_version)
+                ):
+                    distro.set_distributed_publication()
 
             # Unmark old checkpoints if retention is configured
             if self.checkpoint:
@@ -254,7 +267,8 @@ def __exit__(self, exc_type, exc_val, exc_tb):
             # invalidate cache
             if settings.CACHE_ENABLED:
                 base_paths = Distribution.objects.filter(
-                    repository=self.repository_version.repository
+                    models.Q(repository=self.repository_version.repository)
+                    | models.Q(repository_version=self.repository_version)
                 ).values_list("base_path", flat=True)
                 if base_paths:
                     Cache().delete(base_key=cache_key(base_paths))
@@ -800,14 +814,37 @@ def get_fallback_ca(self, path):
         is_not=None,
     )
     def set_distributed_publication(self):
-        """Track the publication being served when a distribution is created or changed."""
+        """
+        Track the publication currently served by this distribution.
+
+        Records a DistributedPublication for the publication this distribution resolves to
+        (directly, or indirectly via its repository/repository_version). Idempotent: does
+        nothing if the active DistributedPublication already points at that publication.
+        """
         detail = self.cast()
         if not detail.SERVE_FROM_PUBLICATION or not retain_distributed_pub_enabled():
             return
         _, _, pub = detail.get_repository_publication_and_version()
         if pub is None:
             return
-        DistributedPublication(distribution=self, publication=pub).save()
+        # Check if this publication is already the active one
+        already_current = DistributedPublication.objects.filter(
+            distribution=self, publication=pub, expires_at__isnull=True
+        ).exists()
+        if already_current:
+            return
+
+        # Check if there's a non-expired DP for this publication that needs reactivation
+        expiring_dp = DistributedPublication.objects.filter(
+            distribution=self, publication=pub, expires_at__isnull=False
+        ).first()
+        if expiring_dp:
+            # Reactivate by clearing expires_at instead of creating a duplicate
+            expiring_dp.expires_at = None
+            expiring_dp.save(skip_hooks=True)  # Avoid triggering cleanup() again
+        else:
+            # No existing DP for this publication, create a new one
+            DistributedPublication(distribution=self, publication=pub).save()
 
     @hook(
         AFTER_UPDATE,

pulpcore/tests/unit/models/test_publication_retention.py

@@ -1,5 +1,7 @@
 import hashlib
 import uuid
+from datetime import timedelta
+from unittest import mock
 
 import pytest
 
@@ -9,6 +11,7 @@
     DistributedPublication,
     PublishedArtifact,
 )
+from pulpcore.app.util import cache_key
 
 from pulp_file.app.models import (
     FileContent,
@@ -50,15 +53,50 @@ def update_dist(dist, repo=UNSET, repover=UNSET, pub=UNSET):
     assert (repo, repover, pub).count(UNSET) == 2, (
         "Exactly one of repo, repover, or pub must be provided"
     )
+    # Clear the other fields to ensure mutual exclusivity
     if repo is not UNSET:
+        dist.publication = None
+        dist.repository_version = None
         dist.repository = repo
     if repover is not UNSET:
+        dist.publication = None
+        dist.repository = None
         dist.repository_version = repover
     if pub is not UNSET:
+        dist.repository = None
+        dist.repository_version = None
         dist.publication = pub
     dist.save()
 
 
+def publish(repo_version, pass_through=True):
+    """
+    Create and complete a publication through the Publication context manager.
+
+    `CreatedResource` is mocked out because it requires a current Task, which isn't set up in
+    unit tests. This still runs the `__exit__` finalization that invalidates caches and records
+    distributed publications.
+    """
+    with mock.patch("pulpcore.app.models.publication.CreatedResource"):
+        with FilePublication.create(repo_version, pass_through=pass_through) as pub:
+            pass
+    return pub
+
+
+def invalidated_base_paths(mock_cache):
+    """Collect the set of base_paths passed to a mocked ``Cache().delete``."""
+    paths = set()
+    for call in mock_cache.return_value.delete.call_args_list:
+        base_key = call.kwargs.get("base_key")
+        if base_key is None and call.args:
+            base_key = call.args[0]
+        if isinstance(base_key, str):
+            paths.add(base_key)
+        elif base_key is not None:
+            paths.update(base_key)
+    return paths
+
+
 def create_version(repo, add=None, remove=None):
     """
     Create a RepositoryVersion adding and/or removing content by path.
@@ -137,6 +175,65 @@ def test_unaffected_when_older_repository_version_deleted(self, db):
         v1.delete()
         assert DistributedPublication.objects.filter(distribution=dist).count() == 1
 
+    def test_created_when_new_publication_for_distributed_repository(self, db):
+        # A distribution serving a repository directly indirectly distributes the latest
+        # publication. Creating a new publication should record it as distributed.
+        repo = FileRepository.objects.create(name=f"repo-{uuid.uuid4().hex[:8]}")
+        version = create_version(repo, add=["some-file.txt"])
+        dist = dist_factory(repo=repo)
+        pub = publish(version)
+        assert DistributedPublication.objects.filter(distribution=dist, publication=pub).exists()
+
+    def test_created_when_new_publication_for_distributed_repository_version(self, db):
+        # A distribution serving a repository_version (with SERVE_FROM_PUBLICATION) indirectly
+        # distributes the latest publication of that version. Creating a new publication should
+        # record it as distributed, just like the repository case above.
+        repo = FileRepository.objects.create(name=f"repo-{uuid.uuid4().hex[:8]}")
+        version = create_version(repo, add=["some-file.txt"])
+        dist = dist_factory(repover=version)
+        pub = publish(version)
+        assert DistributedPublication.objects.filter(distribution=dist, publication=pub).exists()
+
+    def test_reuses_non_expired_distributed_publication(self, db):
+        # If a distribution is updated to serve a publication for which a DistributedPublication
+        # already exists (but is expiring, not active), that DP should be reactivated instead of
+        # creating a duplicate.
+        pub1 = pub_factory()
+        pub2 = pub_factory()
+        dist = dist_factory(pub=pub1)
+        # dist→pub1: DP(pub1, expires_at=NULL)
+        assert (
+            DistributedPublication.objects.filter(
+                distribution=dist, publication=pub1, expires_at__isnull=True
+            ).count()
+            == 1
+        )
+
+        # Switch to pub2: DP(pub1) gets expires_at set, DP(pub2, expires_at=NULL) created
+        update_dist(dist, pub=pub2)
+        dp_pub1_old = DistributedPublication.objects.get(distribution=dist, publication=pub1)
+        assert dp_pub1_old.expires_at is not None
+        assert (
+            DistributedPublication.objects.filter(
+                distribution=dist, publication=pub2, expires_at__isnull=True
+            ).count()
+            == 1
+        )
+
+        # Switch back to pub1: should reactivate the existing DP(pub1), not create a new one
+        dist.refresh_from_db()
+        update_dist(dist, pub=pub1)
+        all_dps_pub1 = DistributedPublication.objects.filter(distribution=dist, publication=pub1)
+        assert all_dps_pub1.count() == 1, (
+            f"Should reuse existing DP, not create duplicate. "
+            f"Found {all_dps_pub1.count()} DPs for pub1"
+        )
+        dp_pub1_reactivated = all_dps_pub1.first()
+        assert dp_pub1_reactivated.pk == dp_pub1_old.pk, (
+            "Should be the same DP instance, reactivated"
+        )
+        assert dp_pub1_reactivated.expires_at is None, "Should clear expires_at when reactivating"
+
 
 @pytest.mark.django_db
 class TestGetFallbackCa:
@@ -210,3 +307,89 @@ def version_without_content(self, version_with_content):
     @pytest.fixture
     def expected_ca(self, version_with_content):
         return ContentArtifact.objects.get(relative_path=self.content_path)
+
+
+@pytest.mark.django_db
+class TestCacheInvalidationOnPublicationCreate:
+    """
+    Creating a new publication changes the content served by every distribution that
+    indirectly distributes that publication's repository/repository_version. The cache
+    must be invalidated for all of them, not just the ones with a direct ``repository`` FK.
+    """
+
+    def test_invalidates_repository_distribution(self, settings):
+        settings.CACHE_ENABLED = True
+        repo = FileRepository.objects.create(name=f"repo-{uuid.uuid4().hex[:8]}")
+        version = create_version(repo, add=["some-file.txt"])
+        dist = dist_factory(repo=repo)
+        with mock.patch("pulpcore.app.models.publication.Cache") as mock_cache:
+            publish(version)
+        assert cache_key(dist.base_path) in invalidated_base_paths(mock_cache)
+
+    def test_invalidates_repository_version_distribution(self, settings):
+        settings.CACHE_ENABLED = True
+        repo = FileRepository.objects.create(name=f"repo-{uuid.uuid4().hex[:8]}")
+        version = create_version(repo, add=["some-file.txt"])
+        dist = dist_factory(repover=version)
+        with mock.patch("pulpcore.app.models.publication.Cache") as mock_cache:
+            publish(version)
+        assert cache_key(dist.base_path) in invalidated_base_paths(mock_cache)
+
+
+@pytest.mark.django_db
+class TestCacheInvalidationOnPublicationDelete:
+    """
+    Deleting the publication currently served by a distribution changes what that distribution
+    serves, so its cache must be invalidated -- including distributions that serve the
+    publication indirectly through their repository_version.
+    """
+
+    def test_invalidates_repository_distribution(self, settings):
+        settings.CACHE_ENABLED = True
+        repo = FileRepository.objects.create(name=f"repo-{uuid.uuid4().hex[:8]}")
+        version = create_version(repo, add=["some-file.txt"])
+        pub = pub_factory(version, pass_through=True)
+        dist = dist_factory(repo=repo)
+        with mock.patch("pulpcore.app.models.publication.Cache") as mock_cache:
+            pub.delete()
+        assert cache_key(dist.base_path) in invalidated_base_paths(mock_cache)
+
+    def test_invalidates_repository_version_distribution(self, settings):
+        settings.CACHE_ENABLED = True
+        repo = FileRepository.objects.create(name=f"repo-{uuid.uuid4().hex[:8]}")
+        version = create_version(repo, add=["some-file.txt"])
+        pub = pub_factory(version, pass_through=True)
+        dist = dist_factory(repover=version)
+        with mock.patch("pulpcore.app.models.publication.Cache") as mock_cache:
+            pub.delete()
+        assert cache_key(dist.base_path) in invalidated_base_paths(mock_cache)
+
+    def test_does_not_invalidate_repository_version_distribution_for_other_version(self, settings):
+        # Deleting a publication of a different version must not invalidate a distribution that
+        # serves an unrelated repository_version.
+        settings.CACHE_ENABLED = True
+        repo = FileRepository.objects.create(name=f"repo-{uuid.uuid4().hex[:8]}")
+        version1 = create_version(repo, add=["v1.txt"])
+        version2 = create_version(repo, add=["v2.txt"])
+        pub2 = pub_factory(version2, pass_through=True)
+        dist = dist_factory(repover=version1)
+        with mock.patch("pulpcore.app.models.publication.Cache") as mock_cache:
+            pub2.delete()
+        assert cache_key(dist.base_path) not in invalidated_base_paths(mock_cache)
+
+    def test_does_not_invalidate_when_deleting_superseded_publication(self, settings):
+        # Deleting an older (non-latest) publication of the served version must not invalidate the
+        # repository_version distribution, which still serves the newer publication.
+        settings.CACHE_ENABLED = True
+        repo = FileRepository.objects.create(name=f"repo-{uuid.uuid4().hex[:8]}")
+        version = create_version(repo, add=["some-file.txt"])
+        old_pub = pub_factory(version, pass_through=True)
+        new_pub = pub_factory(version, pass_through=True)
+        # Force deterministic ordering (pulp_created is auto_now_add).
+        FilePublication.objects.filter(pk=old_pub.pk).update(
+            pulp_created=new_pub.pulp_created - timedelta(seconds=1)
+        )
+        dist = dist_factory(repover=version)
+        with mock.patch("pulpcore.app.models.publication.Cache") as mock_cache:
+            old_pub.delete()
+        assert cache_key(dist.base_path) not in invalidated_base_paths(mock_cache)