Fix cache and DistributedPublication for repository_version distributions

Distributions serving via `repository_version` were missed by cache invalidation
and DistributedPublication tracking when publications were created or deleted.

Also fixes DistributedPublication to reactivate (clear `expires_at`) when switching
back to a recently-superseded publication, instead of creating a duplicate record.

closes #7993
Assisted-By: Claude Opus 4.6

Author: dralley

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,46 @@ 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()
+
+        # Fast path: check if already active without locking (handles most calls)
+        if DistributedPublication.objects.filter(
+            distribution=self, publication=pub, expires_at__isnull=True
+        ).exists():
+            return
+
+        # Slow path: need to create or reactivate under lock
+        with transaction.atomic():
+            # Lock the DP for this (distribution, publication) pair if it exists
+            # At most one row exists per pair (either active or expiring, not both)
+            dp = (
+                DistributedPublication.objects.filter(distribution=self, publication=pub)
+                .select_for_update()
+                .first()
+            )
+
+            if dp is None:
+                # No DP exists, create one
+                DistributedPublication(distribution=self, publication=pub).save()
+            elif dp.expires_at is None:
+                # Already active (race between fast path and here)
+                return
+            else:
+                # Expiring DP exists, reactivate by clearing expires_at
+                dp.expires_at = None
+                dp.save(skip_hooks=True)
 
     @hook(
         AFTER_UPDATE,