Skip /tags/list enumeration for digest-only sync

CHANGES/2474.feature

@@ -0,0 +1 @@
+Added tag list bypass optimization for digest-based sync. When all entries in a remote's `includes` field are sha256 digests, the sync pipeline skips the expensive `/tags/list` enumeration and fetches manifests directly by digest.

pulp_container/app/tasks/sync_stages.py

@@ -3,6 +3,7 @@
 import hashlib
 import json
 import logging
+import re
 from collections import defaultdict
 from urllib.parse import urljoin, urlparse, urlunparse
 
@@ -52,12 +53,13 @@ class ContainerFirstStage(Stage):
 
     """
 
-    def __init__(self, remote, signed_only):
+    def __init__(self, remote, signed_only, mirror=False):
         """Initialize the stage."""
         super().__init__()
         self.remote = remote
         self.deferred_download = self.remote.policy != Remote.IMMEDIATE
         self.signed_only = signed_only
+        self.mirror = mirror
 
         self.tag_dcs = []
         self.manifest_list_dcs = []
@@ -66,6 +68,9 @@ def __init__(self, remote, signed_only):
         self._synced_digests = defaultdict(list)
         self._full_tag_list = []
         self._cosign_tags = []
+        self._bypass_taglist = False
+        self._probed_cosign_digests = set()
+        self._cosign_probe_semaphore = asyncio.Semaphore(20)
 
     async def _download_manifest_data(self, manifest_url):
         downloader = self.remote.get_downloader(url=manifest_url)
@@ -116,12 +121,99 @@ async def _check_for_existing_manifest(self, head_manifest_task):
 
         return content_data, raw_text_data, response, original_reference
 
+    def _can_bypass_taglist(self):
+        """
+        Check if we can safely bypass /tags/list enumeration.
+
+        Returns True only if all includes are explicit sha256 digests
+        and we are not in mirror mode (which needs the full list to detect removals).
+        """
+        includes = self.remote.includes or []
+
+        if not includes:
+            return False
+
+        if self.mirror:
+            return False
+
+        return all(re.fullmatch(r"sha256:[0-9a-f]{64}", entry) for entry in includes)
+
+    async def _tag_exists(self, tag_name):
+        """Check if a tag exists upstream via lightweight HEAD request."""
+        relative_url = "/v2/{name}/manifests/{tag}".format(
+            name=self.remote.namespaced_upstream_name, tag=tag_name
+        )
+        manifest_url = urljoin(self.remote.url, relative_url)
+        downloader = self.remote.get_downloader(url=manifest_url)
+
+        try:
+            await downloader.run(extra_data={"headers": V2_ACCEPT_HEADERS, "http_method": "head"})
+            return True
+        except aiohttp.ClientResponseError as exc:
+            if exc.status == 404:
+                return False
+            raise
+
+    async def _ensure_cosign_tags(self, digests):
+        """
+        Populate _cosign_tags by probing for cosign companion tags via HEAD requests.
+
+        Only used when _bypass_taglist is True. Results are cached per digest
+        so repeated calls (e.g. from _has_cosign_signature and then the post-sync
+        companion pass) do not re-probe.
+        """
+        if not self._bypass_taglist:
+            return
+
+        pending = sorted(set(digests) - self._probed_cosign_digests)
+        candidates = []
+
+        for digest in pending:
+            if not digest.startswith("sha256:"):
+                continue
+
+            stem = digest.replace("sha256:", "sha256-", 1)
+            candidates.extend(f"{stem}{suffix}" for suffix in (*COSIGN_TAG_SUFFIXES, ""))
+
+        candidates = filter_resources(candidates, [], self.remote.excludes or [])
+
+        async def probe(tag):
+            async with self._cosign_probe_semaphore:
+                return tag if await self._tag_exists(tag) else None
+
+        results = await asyncio.gather(*(probe(tag) for tag in candidates))
+        self._cosign_tags.extend(tag for tag in results if tag is not None)
+        self._probed_cosign_digests.update(pending)
+
     async def run(self):
         """
         ContainerFirstStage.
         """
         signature_source = await self.get_signature_source()
 
+        self._bypass_taglist = self._can_bypass_taglist()
+        if self._bypass_taglist:
+            digest_list = list(self.remote.includes)
+            log.info(
+                "Bypassing /tags/list enumeration - syncing %d explicit digests directly",
+                len(digest_list),
+            )
+            await self._process_manifests(digest_list, signature_source, "Processing Manifests")
+
+            await self._ensure_cosign_tags(list(self._synced_digests))
+            companion_tags = self._find_cosign_companion_tags()
+            if companion_tags:
+                log.info(
+                    "Found %d cosign companion tag(s) for synced manifests",
+                    len(companion_tags),
+                )
+                await self._process_manifests(
+                    companion_tags,
+                    signature_source,
+                    "Processing Cosign Companion Tags",
+                )
+            return
+
         async with ProgressReport(
             message="Downloading tag list", code="sync.downloading.tag_list", total=1
         ) as pb:
@@ -300,6 +392,7 @@ async def _process_manifests(self, manifests, signature_source, msg):
 
     async def _has_cosign_signature(self, digest):
         """Check if a digest has a cosign signature."""
+        await self._ensure_cosign_tags([digest])
         cosign_digest = digest.replace("sha256:", "sha256-")
         if f"{cosign_digest}.sig" in self._cosign_tags:
             return True

pulp_container/app/tasks/synchronize.py

@@ -37,7 +37,7 @@ def synchronize(remote_pk, repository_pk, mirror, signed_only):
     remote = ContainerRemote.objects.get(pk=remote_pk)
     repository = ContainerRepository.objects.get(pk=repository_pk)
     log.info("Synchronizing: repository={r} remote={p}".format(r=repository.name, p=remote.name))
-    first_stage = ContainerFirstStage(remote, signed_only)
+    first_stage = ContainerFirstStage(remote, signed_only, mirror=mirror)
     dv = ContainerDeclarativeVersion(first_stage, repository, mirror)
     rv = dv.create()
     if rv: