Optimize sync for repositories with specific tag lists

CHANGES/2474.feature

@@ -0,0 +1 @@
+Optimize sync performance for repositories with specific tag lists by bypassing expensive /tags/list pagination when include_tags contains only non-wildcard references. Reduces sync time from minutes to seconds for deep repositories (50K+ tags) when syncing specific digests. Cosign companion tags are discovered via HEAD probing.

pulp_container/app/migrations/0046_containerremote_auto_discover_cosign.py

@@ -0,0 +1,16 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("container", "0045_alter_manifest_compressed_image_size"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="containerremote",
+            name="auto_discover_cosign",
+            field=models.BooleanField(default=True),
+        ),
+    ]

pulp_container/app/models.py

@@ -490,6 +490,7 @@ class ContainerRemote(Remote, AutoAddObjPermsMixin):
     include_tags = fields.ArrayField(models.TextField(null=True), null=True)
     exclude_tags = fields.ArrayField(models.TextField(null=True), null=True)
     sigstore = models.TextField(null=True)
+    auto_discover_cosign = models.BooleanField(default=True)
 
     TYPE = "container"
 

pulp_container/app/tasks/sync_stages.py

@@ -52,12 +52,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 = []
@@ -111,12 +112,90 @@ async def _check_for_existing_manifest(self, download_tag):
 
         return content_data, raw_text_data, response
 
+    def _can_bypass_taglist(self):
+        """
+        Check if we can safely bypass /tags/list enumeration.
+
+        Returns True only if:
+        - include_tags contains ONLY sha256 digests (not tag names)
+        - exclude_tags is empty OR won't match any includes (harmless)
+        - No wildcards in include_tags (need exact refs, not patterns)
+        - Not in mirror mode (would need full list to detect removals)
+
+        IMPORTANT: We can only bypass when syncing digests, not tag names.
+        When syncing tag names (e.g., "manifest_a"), we need /tags/list to
+        get their digests for cosign companion discovery.
+        """
+        include_tags = self.remote.include_tags or []
+        exclude_tags = self.remote.exclude_tags or []
+
+        if not include_tags:
+            return False
+
+        # CRITICAL: Only bypass if ALL includes are sha256 digests
+        # Tag names need /tags/list to resolve their digests
+        if not all(tag.startswith("sha256:") for tag in include_tags):
+            return False
+
+        # If excludes exist, check if they're harmless (won't match our includes)
+        if exclude_tags:
+            # Satellite often adds '*-source' exclude which doesn't match sha256 digests
+            harmless_excludes = all(
+                exclude.endswith("-source")
+                or (
+                    exclude.startswith("*")
+                    and not any(tag.endswith(exclude.lstrip("*")) for tag in include_tags)
+                )
+                for exclude in exclude_tags
+            )
+            if not harmless_excludes:
+                return False
+
+        if self.mirror:
+            return False
+
+        wildcard_chars = ["*", "?", "["]
+        includes_str = "".join(include_tags)
+        if any(char in includes_str for char in wildcard_chars):
+            return False
+
+        return True
+
     async def run(self):
         """
         ContainerFirstStage.
         """
         signature_source = await self.get_signature_source()
 
+        # Optimization: if syncing specific refs (no wildcards, no excludes),
+        # skip expensive /tags/list enumeration and sync them directly
+        if self._can_bypass_taglist():
+            log.info(
+                "Bypassing /tags/list enumeration - syncing %d explicit references directly",
+                len(self.remote.include_tags),
+            )
+            await self._process_tags(
+                self.remote.include_tags, signature_source, msg="Processing Manifests"
+            )
+
+            # Auto-discover cosign companion tags if enabled
+            if getattr(self.remote, "auto_discover_cosign", True):
+                log.info("Auto-discovering cosign companion tags via HEAD probing")
+                companion_tags = await self._discover_cosign_companions_without_taglist(
+                    self._synced_digests
+                )
+                if companion_tags:
+                    log.info(
+                        "Found %d cosign companion tag(s) for synced manifests",
+                        len(companion_tags),
+                    )
+                    await self._process_tags(
+                        companion_tags,
+                        signature_source,
+                        msg="Processing Cosign Companion Tags",
+                    )
+            return
+
         async with ProgressReport(
             message="Downloading tag list", code="sync.downloading.tag_list", total=1
         ) as pb:
@@ -150,6 +229,62 @@ async def run(self):
                     companion_tags, signature_source, msg="Processing Cosign Companion Tags"
                 )
 
+    async def _tag_exists(self, tag_name):
+        """Check if a tag exists via lightweight HEAD request."""
+        from urllib.parse import urljoin
+
+        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 Exception:
+            return False
+
+    async def _discover_cosign_companions_without_taglist(self, synced_digests):
+        """
+        Discover cosign companion tags by probing expected patterns via HEAD requests.
+
+        Used when bypassing /tags/list to avoid expensive enumeration.
+        Probes for known cosign patterns:
+        - V2: sha256-<digest>.sig, sha256-<digest>.att, sha256-<digest>.sbom
+        - V3: sha256-<digest> (71 chars, verified via manifest check)
+        """
+        companion_tags = []
+        semaphore = asyncio.Semaphore(20)  # Limit concurrent probes
+
+        async def probe_tag(tag):
+            async with semaphore:
+                if await self._tag_exists(tag):
+                    return tag
+                return None
+
+        # Build list of potential cosign tags to probe
+        candidates = []
+        for digest in synced_digests:
+            # digest format: "sha256:abc123..."
+            if not digest.startswith("sha256:"):
+                continue
+            digest_hex = digest.split(":", 1)[1]
+
+            # V2 cosign patterns
+            candidates.append(f"sha256-{digest_hex}.sig")
+            candidates.append(f"sha256-{digest_hex}.att")
+            candidates.append(f"sha256-{digest_hex}.sbom")
+
+            # V3 cosign pattern (71 chars total)
+            candidates.append(f"sha256-{digest_hex}")
+
+        # Probe all candidates concurrently
+        results = await asyncio.gather(*[probe_tag(tag) for tag in candidates])
+        companion_tags = [tag for tag in results if tag]
+
+        return companion_tags
+
     def _find_cosign_companion_tags(self):
         """Find cosign companion tags for synced digests."""
         companion_tags = []

pulp_container/app/tasks/synchronize.py

@@ -36,7 +36,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)
     return dv.create()
 

pulp_container/tests/unit/test_sync_stages.py

@@ -152,5 +152,125 @@ async def test_has_cosign_signature_false_when_no_cosign_tags(self):
         self.assertFalse(await self.stage._has_cosign_signature(digest))
 
 
+class TestBypassTaglistOptimization(unittest.IsolatedAsyncioTestCase):
+    """Test bypass logic for skipping /tags/list enumeration."""
+
+    def setUp(self):
+        remote = MagicMock()
+        remote.policy = MagicMock()
+        remote.namespaced_upstream_name = "library/test"
+        remote.url = "https://registry.example/"
+        remote.get_downloader = MagicMock()
+        remote.include_tags = None
+        remote.exclude_tags = None
+        remote.auto_discover_cosign = True
+
+        self.stage = ContainerFirstStage(remote=remote, signed_only=False, mirror=False)
+
+    def test_can_bypass_with_specific_digests_only(self):
+        """Bypass activates when include_tags contains only sha256 digests."""
+        self.stage.remote.include_tags = [
+            "sha256:abc123",
+            "sha256:def456",
+        ]
+        self.stage.remote.exclude_tags = None
+        self.assertTrue(self.stage._can_bypass_taglist())
+
+    def test_cannot_bypass_without_includes(self):
+        """Bypass does not activate when include_tags is empty."""
+        self.stage.remote.include_tags = None
+        self.stage.remote.exclude_tags = None
+        self.assertFalse(self.stage._can_bypass_taglist())
+
+    def test_cannot_bypass_with_tag_names(self):
+        """Bypass does not activate when include_tags contains tag names (not digests)."""
+        self.stage.remote.include_tags = ["manifest_a", "latest"]
+        self.stage.remote.exclude_tags = None
+        self.assertFalse(self.stage._can_bypass_taglist())
+
+    def test_cannot_bypass_with_mixed_digests_and_tag_names(self):
+        """Bypass does not activate when include_tags mixes digests and tag names."""
+        self.stage.remote.include_tags = ["sha256:abc123", "manifest_a"]
+        self.stage.remote.exclude_tags = None
+        self.assertFalse(self.stage._can_bypass_taglist())
+
+    def test_cannot_bypass_with_wildcards(self):
+        """Bypass does not activate when include_tags contains wildcards."""
+        self.stage.remote.include_tags = ["v4.0*", "sha256:abc123"]
+        self.stage.remote.exclude_tags = None
+        self.assertFalse(self.stage._can_bypass_taglist())
+
+    def test_cannot_bypass_in_mirror_mode(self):
+        """Bypass does not activate in mirror mode."""
+        self.stage.mirror = True
+        self.stage.remote.include_tags = ["sha256:abc123"]
+        self.stage.remote.exclude_tags = None
+        self.assertFalse(self.stage._can_bypass_taglist())
+
+    def test_can_bypass_with_harmless_excludes(self):
+        """Bypass activates when excludes won't match sha256 includes."""
+        self.stage.remote.include_tags = ["sha256:abc123", "sha256:def456"]
+        self.stage.remote.exclude_tags = ["*-source"]
+        self.assertTrue(self.stage._can_bypass_taglist())
+
+    def test_cannot_bypass_with_harmful_excludes(self):
+        """Bypass does not activate when excludes could match digest includes."""
+        self.stage.remote.include_tags = ["sha256:abc123", "sha256:def456"]
+        self.stage.remote.exclude_tags = ["sha256:*"]
+        self.assertFalse(self.stage._can_bypass_taglist())
+
+    def test_cannot_bypass_with_overlapping_excludes(self):
+        """Bypass does not activate when exclude pattern matches an include."""
+        self.stage.remote.include_tags = ["sha256:abc123", "v4.0-source"]
+        self.stage.remote.exclude_tags = ["*-source"]
+        self.assertFalse(self.stage._can_bypass_taglist())
+
+    async def test_tag_exists_returns_true_on_200(self):
+        """_tag_exists returns True when HEAD request succeeds."""
+        downloader = MagicMock()
+        mock_result = AsyncMock()
+        mock_result.status_code = 200
+        downloader.run = AsyncMock(return_value=mock_result)
+        self.stage.remote.get_downloader.return_value = downloader
+
+        result = await self.stage._tag_exists("test-tag")
+        self.assertTrue(result)
+
+    async def test_tag_exists_returns_false_on_404(self):
+        """_tag_exists returns False when tag doesn't exist."""
+        downloader = MagicMock()
+        downloader.run = AsyncMock(side_effect=Exception("404"))
+        self.stage.remote.get_downloader.return_value = downloader
+
+        result = await self.stage._tag_exists("missing-tag")
+        self.assertFalse(result)
+
+    async def test_discover_cosign_companions_probes_variants(self):
+        """Discover cosign companions probes .sig, .att, .sbom variants."""
+        synced_digests = {"sha256:abc123"}
+        self.stage._synced_digests = synced_digests
+
+        # Mock _tag_exists to return True for .sig variant only
+        async def mock_tag_exists(tag):
+            return tag == "sha256-abc123.sig"
+
+        self.stage._tag_exists = AsyncMock(side_effect=mock_tag_exists)
+
+        companions = await self.stage._discover_cosign_companions_without_taglist(synced_digests)
+
+        self.assertEqual(len(companions), 1)
+        self.assertEqual(companions[0], "sha256-abc123.sig")
+
+    async def test_discover_cosign_companions_returns_empty_when_none_exist(self):
+        """Discover returns empty list when no companion tags exist."""
+        synced_digests = {"sha256:abc123"}
+        self.stage._synced_digests = synced_digests
+        self.stage._tag_exists = AsyncMock(return_value=False)
+
+        companions = await self.stage._discover_cosign_companions_without_taglist(synced_digests)
+
+        self.assertEqual(len(companions), 0)
+
+
 if __name__ == "__main__":
     unittest.main()