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

Same fix as nxtgauge-frontend-solid: registry_prune.py sorted candidate
tags by the manifest GET response's Date header (the moment of the
request, not the image's actual build time), making the "keep newest N"
sort effectively random whenever multiple tags are touched in the same
prune run - which happens on every single run, since the tag just
pushed by this same build is always one of the candidates. That let the
prune step delete the image a run had just built, before gitops even
got a chance to reference it.

Now reads the real "created" timestamp from the image's config blob and
always protects the current run's own SHA from deletion regardless of
sort order.

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)

.forgejo/workflows/build.yaml

@@ -87,6 +87,7 @@ jobs:
           REGISTRY_NAMESPACE: ${{ secrets.REGISTRY_NAMESPACE || 'ashwin' }}
           REGISTRY_USERNAME: ${{ secrets.REGISTRY_USERNAME }}
           REGISTRY_PASSWORD: ${{ secrets.REGISTRY_PASSWORD }}
+          SHA: ${{ github.sha }}
         run: |
           set -euo pipefail
           python3 .forgejo/scripts/registry_prune.py \
@@ -94,6 +95,7 @@ jobs:
             --repo "$REGISTRY_NAMESPACE/nxtgauge-admin-solid" \
             --username "$REGISTRY_USERNAME" \
             --password "$REGISTRY_PASSWORD" \
+            --protect "$SHA" \
             --keep 2
 
       - name: Update GitOps release