fix: verify GHCR release manifests when tag listing is huge

cc-ci-plan/resolve-images.py

@@ -313,6 +313,39 @@ def _release_fallback_repos(registry: str, repo: str) -> list[tuple[str, str]]:
     return cands
 
 
+def _oci_tag_exists(host: str, repo: str, tag: str) -> bool:
+    """Verify a release candidate is actually published as an image manifest."""
+    hdrs = {"User-Agent": UA,
+            "Accept": "application/vnd.oci.image.index.v1+json, "
+                      "application/vnd.docker.distribution.manifest.list.v2+json, "
+                      "application/vnd.oci.image.manifest.v1+json",
+            **_oci_bearer(host, repo)}
+    url = f"https://{host}/v2/{repo}/manifests/{urllib.parse.quote(tag, safe='')}"
+    try:
+        with urllib.request.urlopen(urllib.request.Request(url, headers=hdrs, method="HEAD"),
+                                    timeout=TIMEOUT) as response:
+            return response.status == 200
+    except urllib.error.HTTPError as e:
+        if e.code == 404:
+            return False
+        raise
+
+
+def _release_manifest_fallback(registry: str, repo: str, current: str) -> tuple[list[str], str] | None:
+    """For huge GHCR registries, use release tags only after checking actual manifests."""
+    for owner, name in _release_fallback_repos(registry, repo):
+        releases = github_release_tags(owner, name)
+        if current not in releases:
+            continue  # incomplete release window; it cannot establish a current baseline
+        candidates = sorted((tag for tag in set(releases)
+                             if shape(tag) == shape(current) and vkey(tag) >= vkey(current)),
+                            key=vkey, reverse=True)
+        for tag in candidates:
+            if _oci_tag_exists(registry, repo, tag):
+                return [current, tag], f"github-releases+registry-manifest:{owner}/{name}"
+    return None
+
+
 _TAG_CACHE: dict[tuple[str, str], tuple[list[str], str | None]] = {}
 
 
@@ -349,8 +382,16 @@ def resolve(ref: str) -> dict:
            "upgrade_available": False}
     tags, err = list_tags(info["registry"], info["repo"])
     if err:
-        out["status"] = f"error: {err}"
-        return out
+        try:
+            fallback = (_release_manifest_fallback(info["registry"], info["repo"], info["tag"])
+                        if "tag listing truncated" in err else None)
+        except Exception as e:  # noqa: BLE001 — a failed fallback must stay unresolved
+            out["status"] = f"error: registry {err}; release fallback {type(e).__name__}: {e}"
+            return out
+        if not fallback:
+            out["status"] = f"error: {err}"
+            return out
+        tags, out["source"] = fallback
     out["tags_seen"] = len(set(tags))
     # INTEGRITY CHECK: the tag we are currently running MUST appear in the listing. If it does not,
     # the listing is incomplete and any "newest" derived from it is a guess — ghcr paginates to tens

cc-ci-plan/test-resolve-images.py

@@ -33,6 +33,23 @@ class TestResolverCoverage(unittest.TestCase):
                 row = resolver.resolve("example:1.2.3")
         self.assertTrue(row["status"].startswith("unverifiable:"))
 
+    def test_huge_ghcr_listing_uses_only_published_release_manifests(self):
+        ref = "ghcr.io/immich-app/immich-machine-learning:v3.3.5"
+        with mock.patch.object(resolver, "list_tags", return_value=([], "RuntimeError: tag listing truncated")), \
+             mock.patch.object(resolver, "github_release_tags",
+                               return_value=["v3.3.6", "v3.3.5"]), \
+             mock.patch.object(resolver, "_oci_tag_exists", side_effect=lambda h, r, t: t == "v3.3.5"):
+            row = resolver.resolve(ref)
+        self.assertEqual(row["status"], "ok")
+        self.assertFalse(row["upgrade_available"])  # release exists; image does not yet
+        self.assertIn("registry-manifest", row["source"])
+        with mock.patch.object(resolver, "list_tags", return_value=([], "RuntimeError: tag listing truncated")), \
+             mock.patch.object(resolver, "github_release_tags",
+                               return_value=["v3.3.6", "v3.3.5"]), \
+             mock.patch.object(resolver, "_oci_tag_exists", return_value=True):
+            upgraded = resolver.resolve(ref)
+        self.assertTrue(upgraded["upgrade_available"])
+
 
 if __name__ == "__main__":
     unittest.main()