Serve pull-through blobs while their version commits

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.

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."""

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

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()