From 2b2fe3f840bdc3768663d5a30f90d87017013ee0 Mon Sep 17 00:00:00 2001 From: Daniel Alley Date: Wed, 19 Aug 2026 22:17:15 -0400 Subject: [PATCH] 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 --- CHANGES/7993.bugfix | 1 + pulpcore/app/models/publication.py | 63 ++++-- .../unit/models/test_publication_retention.py | 183 ++++++++++++++++++ 3 files changed, 234 insertions(+), 13 deletions(-) create mode 100644 CHANGES/7993.bugfix diff --git a/CHANGES/7993.bugfix b/CHANGES/7993.bugfix new file mode 100644 index 00000000000..f16822d048a --- /dev/null +++ b/CHANGES/7993.bugfix @@ -0,0 +1 @@ +Fixed cache invalidation and DistributedPublication tracking for distributions serving via `repository_version` when publications are created or deleted. diff --git a/pulpcore/app/models/publication.py b/pulpcore/app/models/publication.py index 7de105a50b2..6c4c26c997a 100644 --- a/pulpcore/app/models/publication.py +++ b/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, diff --git a/pulpcore/tests/unit/models/test_publication_retention.py b/pulpcore/tests/unit/models/test_publication_retention.py index f16c9b1e036..a5102d91356 100644 --- a/pulpcore/tests/unit/models/test_publication_retention.py +++ b/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)