From 084bed1dfa51589bf53853eba661df487d3d50d0 Mon Sep 17 00:00:00 2001 From: Scanner Date: Sat, 31 Oct 2026 18:28:58 -0700 Subject: [PATCH] Serve pull-through blobs while their version commits The content app could answer 404 for a pull-through blob in two windows while the task adding it to the repository committed: - finalize_new_version cleared the blob from pending_blobs, and that committed before pulpcore committed the version as complete. - get_by_digest read the repository version and looked the blob up in separate statements, so a commit in between hid the blob. Clear pending content in on_new_version, after the version is complete, and read the version and the blob in one REPEATABLE READ transaction. fixes #2626 Assisted-by: Claude Opus 5.5 --- CHANGES/2626.bugfix | 1 + pulp_container/app/models.py | 26 ++- pulp_container/app/registry.py | 61 +++++- .../tests/unit/test_pending_blob_race.py | 191 ++++++++++++++++++ 4 files changed, 270 insertions(+), 9 deletions(-) create mode 100644 CHANGES/2626.bugfix create mode 100644 pulp_container/tests/unit/test_pending_blob_race.py diff --git a/CHANGES/2626.bugfix b/CHANGES/2626.bugfix new file mode 100644 index 000000000..8d2043859 --- /dev/null +++ b/CHANGES/2626.bugfix @@ -0,0 +1 @@ +Fixed the content app answering 404 for a pull-through blob while the task that adds it to the repository commits. diff --git a/pulp_container/app/models.py b/pulp_container/app/models.py index b8d8d0fff..73b8810e5 100644 --- a/pulp_container/app/models.py +++ b/pulp_container/app/models.py @@ -755,7 +755,18 @@ def finalize_new_version(self, new_version): """ remove_duplicates(new_version) validate_repo_version(new_version) - self.remove_pending_content(new_version) + + def on_new_version(self, version): + """ + Remove the content the new version added from the pending blobs and manifests. + + pulpcore calls this once the version is complete and committed, so the content is always + either pending or in the latest complete version, never neither, for the content app. + Args: + version (pulpcore.app.models.RepositoryVersion): The new, complete RepositoryVersion. + """ + super().on_new_version(version) + self.remove_pending_content(version) def remove_pending_content(self, repository_version): """Remove pending blobs and manifests when committing the content to the repository.""" @@ -809,7 +820,18 @@ def finalize_new_version(self, new_version): """ remove_duplicates(new_version) validate_repo_version(new_version) - self.remove_pending_content(new_version) + + def on_new_version(self, version): + """ + Remove the content the new version added from the pending blobs and manifests. + + pulpcore calls this once the version is complete and committed, so the content is always + either pending or in the latest complete version, never neither, for the content app. + Args: + version (pulpcore.app.models.RepositoryVersion): The new, complete RepositoryVersion. + """ + super().on_new_version(version) + self.remove_pending_content(version) def remove_pending_content(self, repository_version): """Remove pending blobs and manifests when committing the content to the repository.""" diff --git a/pulp_container/app/registry.py b/pulp_container/app/registry.py index f776ea7f6..d8314bf0d 100644 --- a/pulp_container/app/registry.py +++ b/pulp_container/app/registry.py @@ -4,6 +4,7 @@ from aiohttp import web from asgiref.sync import sync_to_async from django.core.exceptions import ObjectDoesNotExist +from django.db import connection, transaction from multidict import MultiDict from pulpcore.plugin.content import ArtifactResponse, Handler, PathNotResolved @@ -85,19 +86,19 @@ async def get_by_digest(self, request): digest = "sha256:{digest}".format(digest=request.match_info["digest"]) distribution = await sync_to_async(self._match_distribution)(path, add_trailing_slash=False) await sync_to_async(self._permit)(request, distribution) - repository_version = await sync_to_async(distribution.get_repository_version)() + # Read the repository version and look up the blob in one database snapshot. 'ca' is + # None when the blob was not found, and also for the empty blob, which needs no lookup. + repository_version, ca = await sync_to_async( + self._get_repository_version_and_blob_artifact + )(distribution, digest) if not repository_version: raise PathNotResolved(path) if digest == EMPTY_BLOB: return await Registry._empty_blob() - repository = await repository_version.repository.acast() - pending_blobs = repository.pending_blobs.values_list("pk") - content = repository_version.content | Content.objects.filter(pk__in=pending_blobs) + if ca is None: + raise PathNotResolved(path) try: - ca = await ContentArtifact.objects.select_related("artifact", "content").aget( - content__in=content, relative_path=digest - ) ca_content = await ca.content.acast() media_type = BLOB_CONTENT_TYPE headers = { @@ -113,6 +114,52 @@ async def get_by_digest(self, request): else: return await self._stream_content_artifact(request, web.StreamResponse(), ca) + @staticmethod + def _get_repository_version_and_blob_artifact(distribution, digest): + """ + Read the distribution's repository version and find the blob's ContentArtifact. + + The blob is looked up in the version's content and in the repository's pending blobs. + Both reads run in one REPEATABLE READ transaction, so they see the same snapshot of the + database. Since pulp runs only on PostgreSQL using this PostgreSQL specific feature was + alright. + + Returns: + tuple: (repository_version, content_artifact). + repository_version is the RepositoryVersion the distribution serves, or None if it + serves none; content_artifact is then None too. + content_artifact is None when the blob is in neither place, and when the digest is + the empty blob, which the caller serves without a lookup. + """ + with transaction.atomic(): + # SET TRANSACTION has to come before the transaction's first query. Under the default + # READ COMMITTED, each query would see the database as of its own start instead. + with connection.cursor() as cursor: + cursor.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ") + repository_version = distribution.get_repository_version() + + # No repository version: there is nothing to look the blob up in. + if repository_version is None: + return None, None + + # The empty blob is served by the caller without a ContentArtifact, but only when the + # distribution serves a version, so the version is still returned. + if digest == EMPTY_BLOB: + return repository_version, None + + # The blob may be in the version's content or still pending on the repository. "|" + # joins the two querysets with SQL OR, so the get() below is a single query over both. + repository = repository_version.repository.cast() + pending_blobs = repository.pending_blobs.values_list("pk") + content = repository_version.content | Content.objects.filter(pk__in=pending_blobs) + try: + ca = ContentArtifact.objects.select_related("artifact", "content").get( + content__in=content, relative_path=digest + ) + except ObjectDoesNotExist: + ca = None + return repository_version, ca + @staticmethod async def _empty_blob(): # fmt: off diff --git a/pulp_container/tests/unit/test_pending_blob_race.py b/pulp_container/tests/unit/test_pending_blob_race.py new file mode 100644 index 000000000..a53aee177 --- /dev/null +++ b/pulp_container/tests/unit/test_pending_blob_race.py @@ -0,0 +1,191 @@ +import asyncio +import hashlib +import tempfile +import threading +from unittest.mock import patch +from uuid import uuid4 + +from aiohttp.test_utils import make_mocked_request +from django.db import connection +from django.test import TransactionTestCase + +from pulpcore.plugin.content import PathNotResolved +from pulpcore.plugin.models import Artifact, ContentArtifact + +from pulp_container.app.cache import RegistryContentCache +from pulp_container.app.models import Blob, ContainerDistribution, ContainerRepository +from pulp_container.app.registry import Registry +from pulp_container.constants import EMPTY_BLOB + + +class TestPendingBlobDuringVersionCommit(TransactionTestCase): + """ + The content app serves a pull-through blob while a task commits the version that adds it. + + The blob starts in the repository's pending_blobs, as the registry API leaves it before + redirecting a client to the content app. The task runs in another thread, on its own database + connection, as a worker would. A TransactionTestCase lets each side see the other's commits. + """ + + # A TransactionTestCase empties the database after each test, including the rows migrations + # create (such as the default domain). This restores them for the next test. + serialized_rollback = True + + @classmethod + def setUpClass(cls): + super().setUpClass() + # One event loop for every request, as the content app has. With CACHE_ENABLED, the + # cache around get_by_digest holds an async Redis client that stays bound to the loop it + # first ran in. + cls.event_loop = asyncio.Runner() + + @classmethod + def tearDownClass(cls): + # Close the Redis client's connections inside the loop they belong to, so nothing that + # runs later finds them bound to a closed loop. Every content cache shares the one client; + # it is None when the cache is disabled. + redis = RegistryContentCache().redis + if redis is not None: + cls.event_loop.run(redis.connection_pool.disconnect()) + cls.event_loop.close() + super().tearDownClass() + + def setUp(self): + # A blob with a stored artifact. The registry finds a blob's ContentArtifact by + # relative_path, which is the digest. + data = uuid4().bytes + self.digest = "sha256:" + hashlib.sha256(data).hexdigest() + with tempfile.NamedTemporaryFile() as f: + f.write(data) + f.flush() + artifact = Artifact.init_and_validate(f.name) + artifact.save() + self.blob = Blob.objects.create(digest=self.digest) + ContentArtifact.objects.create( + artifact=artifact, content=self.blob, relative_path=self.digest + ) + + # The repository has the blob only in pending_blobs, not yet in any version: the state + # the registry API leaves on a cold pull. The distribution serves the repository's latest + # version. + self.repository = ContainerRepository.objects.create(name=str(uuid4())) + self.repository.pending_blobs.add(self.blob) + self.distribution = ContainerDistribution.objects.create( + name=self.repository.name, + base_path=self.repository.name, + repository=self.repository, + ) + + def commit_version(self): + """ + Commit a new version adding the blob, as the pull-through task does. + + Runs in its own thread, so it uses its own database connection; that connection is + closed on the way out. + """ + try: + with self.repository.new_version() as version: + version.add_content(Blob.objects.filter(pk=self.blob.pk)) + finally: + connection.close() + + def get_blob(self): + """ + Call Registry.get_by_digest for the blob, as the content app would. + + Returns the response; a 404 raises PathNotResolved, which fails a test that does not + expect it. + """ + + # A real aiohttp request, because with CACHE_ENABLED the cache around get_by_digest looks + # for one among the handler's arguments. match_info is what the content app's route + # captures: the digest without its "sha256:" prefix. The request is made inside the event + # loop that handles it. + async def request_blob(): + base_path = self.distribution.base_path + request = make_mocked_request( + "GET", + f"/pulp/container/{base_path}/blobs/{self.digest}", + match_info={"path": base_path, "digest": self.digest[7:]}, + ) + return await Registry().get_by_digest(request) + + return self.event_loop.run(request_blob()) + + def test_blob_served_while_version_commits(self): + """The task is paused after finalize_new_version, before the version is complete.""" + # Wrap finalize_new_version so the task stops right after it and waits to be resumed. + # The 5 second timeouts make a broken test fail instead of hanging. + paused, resume = threading.Event(), threading.Event() + finalize = ContainerRepository.finalize_new_version + + def finalize_then_pause(repository, new_version): + finalize(repository, new_version) + paused.set() + resume.wait(5) + + with patch.object( + ContainerRepository, + "finalize_new_version", + autospec=True, + side_effect=finalize_then_pause, + ): + # Start the task and wait until it is paused, between finalizing the version and + # committing it as complete. Request the blob in that window, then let the task + # finish whatever the request returned. + task = threading.Thread(target=self.commit_version) + task.start() + self.assertTrue(paused.wait(5)) + try: + self.get_blob() + finally: + resume.set() + task.join() + + # Once the version is complete the blob is no longer pending. + self.assertFalse(self.repository.pending_blobs.exists()) + + def test_blob_served_when_version_commits_between_lookups(self): + """The version commits after the content app has read the repository version.""" + # Wrap get_repository_version so that, once get_by_digest has read the version, a task + # commits a new version adding the blob, and finishes, before get_by_digest looks the + # blob up. + get_repository_version = ContainerDistribution.get_repository_version + + def read_then_commit(distribution): + repository_version = get_repository_version(distribution) + task = threading.Thread(target=self.commit_version) + task.start() + task.join() + return repository_version + + with patch.object( + ContainerDistribution, + "get_repository_version", + autospec=True, + side_effect=read_then_commit, + ): + self.get_blob() + + def test_missing_blob_not_found(self): + """A digest that is neither pending nor in the version is still a 404.""" + self.digest = "sha256:" + "0" * 64 + with self.assertRaises(PathNotResolved): + self.get_blob() + + def test_empty_blob_served(self): + """The empty blob is served without a ContentArtifact when a version is served.""" + self.digest = EMPTY_BLOB + response = self.get_blob() + self.assertEqual(response.headers["Docker-Content-Digest"], EMPTY_BLOB) + + def test_no_repository_version_not_found(self): + """With no repository version served, every blob is a 404, the empty blob included.""" + # A distribution with neither a repository nor a repository version serves no version. + self.distribution.repository = None + self.distribution.save() + for digest in (EMPTY_BLOB, self.digest): + with self.subTest(digest=digest): + self.digest = digest + with self.assertRaises(PathNotResolved): + self.get_blob()