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

registry_prune.py sorted candidate tags by the manifest GET response's
Date header - which is just "now", the moment the prune script made
that request - not when the image was actually built. Every tag it
queries in the same prune run lands within the same second, so the
"sort by age" was effectively random. On run #56 this deleted the tag
this same CI run had just pushed (4efe848, digest 60114dea) seconds
after pushing it, and seconds before the gitops step committed a
deployment pointing at that now-deleted digest - the cluster then
sat in ImagePullBackOff since the referenced image no longer existed.

Now reads the real "created" timestamp from the image's config blob
(resolving through a manifest list/index if the tag is multi-platform),
and always protects the current run's own SHA from deletion regardless
of sort order, as defense in depth against any remaining timestamp edge
cases.

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:
-    """Get the digest (sha256:...) and created time for a tag."""
-    url = f"https://{registry}/v2/{repo}/manifests/{tag}"
+def _get_manifest(registry: str, repo: str, ref: str, username: str, password: str):
+    """GET a manifest by tag or digest. Returns (digest, media_type, body_json) or None."""
+    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", "")
-                return digest, created
+                media_type = response.headers.get("Content-Type", "")
+                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)
-    - Keep newest `keep` tags
+    - Sort by the image's actual build timestamp (newest first)
+    - 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:
-    """Parse HTTP Date header to timestamp."""
-    from email.utils import parsedate_to_datetime
+def parse_image_date(date_str: str) -> float:
+    """Parse an OCI image config's ISO-8601 "created" field to a timestamp."""
+    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-frontend-solid" \
             --username "$REGISTRY_USERNAME" \
             --password "$REGISTRY_PASSWORD" \
+            --protect "$SHA" \
             --keep 2
 
       - name: Update GitOps release