fix(ci): sort registry prune by real image build time, protect current SHA

Same fix as nxtgauge-frontend-solid and nxtgauge-admin-solid: this
script isn't currently wired into this repo's build.yaml, but it's
identical to the version that deleted frontend-solid's just-built image
seconds after pushing it (sorted candidates by request time instead of
actual build time). Fixing it here too so it's safe if/when it gets
wired up, rather than leaving a dormant copy of the same bug.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

.forgejo/scripts/registry_prune.py

@@ -1,19 +1,22 @@
 #!/usr/bin/env python3
 """
-Registry Image Tag Pruner - Keeps only the latest 1 SHA-tag per repository.
+Registry Image Tag Pruner - Keeps only the latest N SHA-tag(s) per repository.
 
 Usage:
     python3 registry_prune.py \
         --registry registry.nxtgauge.com \
         --repo nxtgauge-rust-gateway \
         --username "$REGISTRY_USERNAME" \
-        --password "$REGISTRY_PASSWORD"
+        --password "$REGISTRY_PASSWORD" \
+        --protect "$SHA"
 
 Environment variables can also be used:
     REGISTRY_HOST, REGISTRY_REPO, REGISTRY_USERNAME, REGISTRY_PASSWORD
 
 SHA-like tags are identified by pattern: ^[a-f0-9]{40}$
 Non-SHA tags (e.g., high-performance-latest, main-latest, latest) are NEVER deleted.
+--protect tags (e.g. the SHA this very CI run just built) are NEVER deleted,
+regardless of how they sort - this run's own image must survive its own prune.
 
 Exit code: 0 on success (or if prune fails gracefully), non-zero only on critical error.
 """
@@ -27,16 +30,29 @@ import time
 from urllib.request import Request, urlopen
 from urllib.error import URLError, HTTPError
 
+MANIFEST_ACCEPT = ", ".join([
+    "application/vnd.docker.distribution.manifest.list.v2+json",
+    "application/vnd.docker.distribution.manifest.v2+json",
+    "application/vnd.oci.image.index.v1+json",
+    "application/vnd.oci.image.manifest.v1+json",
+])
+
 
 def parse_args():
     parser = argparse.ArgumentParser(
-        description="Prune Docker registry tags, keeping only the latest SHA tag."
+        description="Prune Docker registry tags, keeping only the latest SHA tag(s)."
     )
     parser.add_argument("--registry", default=os.environ.get("REGISTRY_HOST"))
     parser.add_argument("--repo", default=os.environ.get("REGISTRY_REPO"))
     parser.add_argument("--username", default=os.environ.get("REGISTRY_USERNAME"))
     parser.add_argument("--password", default=os.environ.get("REGISTRY_PASSWORD"))
     parser.add_argument("--keep", type=int, default=1, help="Number of SHA tags to keep (default: 1)")
+    parser.add_argument(
+        "--protect",
+        action="append",
+        default=[],
+        help="Tag that must never be deleted (e.g. the SHA this CI run just built). Repeatable.",
+    )
     return parser.parse_args()
 
 
@@ -76,23 +92,68 @@ def api_request(url: str, method: str, username: str, password: str, data=None,
     return None
 
 
-def get_tag_digest(registry: str, repo: str, tag: str, username: str, password: str) -> tuple[str, str] | None:
+def _get_manifest(registry: str, repo: str, ref: str, username: str, password: str):
-    """Get the digest (sha256:...) and created time for a tag."""
+    """GET a manifest by tag or digest. Returns (digest, media_type, body_json) or None."""
-    url = f"https://{registry}/v2/{repo}/manifests/{tag}"
+    url = f"https://{registry}/v2/{repo}/manifests/{ref}"
     auth = base64.b64encode(f"{username}:{password}".encode()).decode()
 
     for attempt in range(1, 4):
         try:
             req = Request(url, method="GET", headers={
                 "Authorization": f"Basic {auth}",
-                "Accept": "application/vnd.docker.distribution.manifest.v2+json",
+                "Accept": MANIFEST_ACCEPT,
             })
             with urlopen(req, timeout=30) as response:
                 digest = response.headers.get("Docker-Content-Digest", "")
-                created = response.headers.get("Date", "")
+                media_type = response.headers.get("Content-Type", "")
-                return digest, created
+                body = json.loads(response.read())
+                return digest, media_type, body
         except Exception as e:
-            print(f"  [RETRY {attempt}/3] Getting digest for {tag}: {e}")
+            print(f"  [RETRY {attempt}/3] Fetching manifest {ref}: {e}")
+            time.sleep(attempt)
+
+    return None
+
+
+def get_tag_created(registry: str, repo: str, tag: str, username: str, password: str) -> tuple[str, str] | None:
+    """
+    Get the digest and true build timestamp for a tag, by reading the
+    "created" field baked into the image's config blob at build time -
+    NOT the HTTP response's Date header, which only reflects the moment
+    this GET request happened (every tag queried in the same prune run
+    ends up with a near-identical Date, so sorting by it is close to
+    random and can rank a tag pushed seconds ago as the oldest one).
+    """
+    result = _get_manifest(registry, repo, tag, username, password)
+    if result is None:
+        return None
+    tag_digest, media_type, body = result
+
+    # A tag can point at a multi-platform manifest list/index rather than
+    # a single image manifest - descend into the first platform entry to
+    # reach an actual image manifest with a "config" pointer.
+    if "manifests" in body and body["manifests"]:
+        child_digest = body["manifests"][0]["digest"]
+        child = _get_manifest(registry, repo, child_digest, username, password)
+        if child is None:
+            return None
+        _, _, body = child
+
+    config = body.get("config")
+    if not config or "digest" not in config:
+        return None
+
+    blob_url = f"https://{registry}/v2/{repo}/blobs/{config['digest']}"
+    auth = base64.b64encode(f"{username}:{password}".encode()).decode()
+    for attempt in range(1, 4):
+        try:
+            req = Request(blob_url, method="GET", headers={"Authorization": f"Basic {auth}"})
+            with urlopen(req, timeout=30) as response:
+                config_json = json.loads(response.read())
+                created = config_json.get("created", "")
+                return tag_digest, created
+        except Exception as e:
+            print(f"  [RETRY {attempt}/3] Fetching config blob for {tag}: {e}")
             time.sleep(attempt)
 
     return None
@@ -129,19 +190,22 @@ def is_sha_tag(tag: str) -> bool:
     return bool(re.match(r"^[a-f0-9]{40}$", tag))
 
 
-def prune_tags(registry: str, repo: str, username: str, password: str, keep: int = 1) -> bool:
+def prune_tags(registry: str, repo: str, username: str, password: str, keep: int = 1, protect: list[str] = ()) -> bool:
     """
     Main prune logic:
     - List all tags for the repo
     - Filter SHA-like tags
-    - Sort by created date (newest first)
+    - Sort by the image's actual build timestamp (newest first)
-    - Keep newest `keep` tags
+    - Keep newest `keep` tags, plus anything in `protect`
     - Delete older SHA tags by digest
-    - Never delete non-SHA tags
+    - Never delete non-SHA tags or protected tags
     """
     print(f"\n=== Pruning {registry}/{repo} ===")
     print(f"Strategy: Keep {keep} newest SHA tag(s), delete older SHA tags")
-    print(f"Non-SHA tags (e.g., high-performance-latest, main-latest, latest) are preserved\n")
+    print(f"Non-SHA tags (e.g., high-performance-latest, main-latest, latest) are preserved")
+    if protect:
+        print(f"Protected SHA tags (never deleted): {', '.join(protect)}")
+    print()
 
     # Get catalog (list of repos)
     catalog_url = f"https://{registry}/v2/_catalog"
@@ -180,17 +244,17 @@ def prune_tags(registry: str, repo: str, username: str, password: str, keep: int
         print("\n[INFO] No SHA tags to prune")
         return True
 
-    # Get digest and created time for each SHA tag
+    # Get digest and true build time for each SHA tag
     tag_info = []
     for tag in sha_tags:
-        result = get_tag_digest(registry, repo, tag, username, password)
+        result = get_tag_created(registry, repo, tag, username, password)
         if result:
             digest, created = result
             tag_info.append({
                 "tag": tag,
                 "digest": digest,
                 "created": created,
-                "timestamp": parse_http_date(created) if created else 0,
+                "timestamp": parse_image_date(created) if created else 0,
             })
         time.sleep(0.1)  # Be nice to the registry
 
@@ -201,17 +265,19 @@ def prune_tags(registry: str, repo: str, username: str, password: str, keep: int
     # Sort by timestamp (newest first)
     tag_info.sort(key=lambda x: x["timestamp"], reverse=True)
 
-    print(f"\nSHA tags sorted by age (newest first):")
+    print(f"\nSHA tags sorted by build time (newest first):")
     for i, info in enumerate(tag_info):
-        marker = " [KEEP]" if i < keep else " [DELETE]"
+        is_protected = info["tag"] in protect
+        will_keep = i < keep or is_protected
+        marker = " [KEEP]" + (" (protected)" if is_protected else "") if will_keep else " [DELETE]"
         print(f"  {i+1}. {info['tag']} ({info['created'] or 'unknown date'}){marker}")
 
-    # Delete older SHA tags
+    # Delete older, unprotected SHA tags
     deleted_count = 0
     kept_count = 0
 
     for i, info in enumerate(tag_info):
-        if i < keep:
+        if i < keep or info["tag"] in protect:
             print(f"\n[KEEP] {info['tag']}")
             kept_count += 1
             continue
@@ -233,11 +299,11 @@ def prune_tags(registry: str, repo: str, username: str, password: str, keep: int
     return True
 
 
-def parse_http_date(date_str: str) -> float:
+def parse_image_date(date_str: str) -> float:
-    """Parse HTTP Date header to timestamp."""
+    """Parse an OCI image config's ISO-8601 "created" field to a timestamp."""
-    from email.utils import parsedate_to_datetime
+    from datetime import datetime
     try:
-        return parsedate_to_datetime(date_str).timestamp()
+        return datetime.fromisoformat(date_str.replace("Z", "+00:00")).timestamp()
     except Exception:
         return 0
 
@@ -261,7 +327,7 @@ def main():
     print(f"Username: {username}")
 
     try:
-        success = prune_tags(registry, repo, username, password, args.keep)
+        success = prune_tags(registry, repo, username, password, args.keep, args.protect)
         if success:
             print("\n[OK] Prune completed successfully")
             sys.exit(0)