Fix an edge case where cache wasn't flushed

Make sure we flush the cache (and update DistributedPublication) for
publications which were indirectly served via
Distribution.repository_version, rather than directly.

closes #7993
Assisted-By: Claude Opus 4.6

Author: dralley

CHANGES/7993.bugfix

@@ -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,13 +814,24 @@ 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
+        already_current = DistributedPublication.objects.filter(
+            distribution=self, publication=pub, expires_at__isnull=True
+        ).exists()
+        if already_current:
+            return
         DistributedPublication(distribution=self, publication=pub).save()
 
     @hook(

pulpcore/app/models/publication.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,
@@ -59,6 +62,34 @@ def update_dist(dist, repo=UNSET, repover=UNSET, pub=UNSET):
     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 +168,25 @@ 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()
+
 
 @pytest.mark.django_db
 class TestGetFallbackCa:
@@ -210,3 +260,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)