From 82503598c27cc378431838656f810c8538563c2a Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Wed, 12 Aug 2026 16:17:20 -0500 Subject: [PATCH] =?UTF-8?q?registry-prune:=20pure=20PDS=20path=20=E2=80=94?= =?UTF-8?q?=20delete=20the=20records=20atcr=20indexes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit atcr has no registry-API manifest DELETE (405 even correctly scoped), so the registry-API path is gone entirely. the script now inventories io.atcr.tag/io.atcr.manifest on our PDS (public reads), scopes strictly to repository stream|stream/cache (the collections also hold zds images), keeps everything reachable from receipted tags including index children resolved through the raw manifest blob, and deletes the rest via authenticated applyWrites. dry run against the live PDS: 6 stale full-sha tags + their 6 manifests doomed, all 16 receipted tags kept, zds untouched. Co-Authored-By: Claude Fable 5 --- .tangled/workflows/registry-prune.yml | 14 +- scripts/registry-prune | 296 +++++++++++++++----------- 2 files changed, 173 insertions(+), 137 deletions(-) diff --git a/.tangled/workflows/registry-prune.yml b/.tangled/workflows/registry-prune.yml index d4067eb..7787ee1 100644 --- a/.tangled/workflows/registry-prune.yml +++ b/.tangled/workflows/registry-prune.yml @@ -1,12 +1,10 @@ # Manual registry housekeeping: prune atcr.io/zat.dev/stream to the tags -# receipts still admit, and drop the kaniko cache repo entirely. Runs in CI -# because that's where the registry credential (ATCR_APP_PASSWORD) lives — -# never on a workstation keychain. -# -# Note: while the account is over quota the token service denies write -# scopes; the script falls back to basic auth for DELETE. If atcr denies -# deletion entirely, the run's log is the evidence to bring to the tangled -# folks. +# receipts still admit. atcr has no registry-API manifest DELETE (405), so +# the script deletes the io.atcr.tag / io.atcr.manifest records on our own +# PDS — the index atcr serves from. That path is not quota-blocked. Runs in +# CI because that's where the credential (ATCR_APP_PASSWORD, an atproto app +# password) lives — never on a workstation keychain. Quota recomputes via +# atcr's server-side GC within ~24h of deletion. when: - event: ["manual"] diff --git a/scripts/registry-prune b/scripts/registry-prune index e8485ef..e811d48 100755 --- a/scripts/registry-prune +++ b/scripts/registry-prune @@ -1,148 +1,186 @@ #!/usr/bin/env python3 -"""Inspect (and attempt to prune) atcr.io/zat.dev/stream. - -STATUS 2026-08-09: atcr does NOT implement manifest deletion — every -`DELETE /v2//manifests/` returns 405 UNSUPPORTED, even with a -correctly scoped token. This script is therefore an INVENTORY tool; the way -images actually get deleted is by removing their `io.atcr.manifest` / -`io.atcr.tag` records from your own PDS, which is what atcr indexes: - - com.atproto.repo.applyWrites (collection=io.atcr.manifest, rkey=) - -That path is not blocked by quota and is what cleared 49 stale tags and the -whole stream/cache repo. Quota is recomputed by a server-side GC roughly -daily, so the usage figure lags deletion by up to ~24h. - -Scope note, worth keeping: `delete` must be requested ALONE. atcr denies -pull/push while over quota, so "pull,push,delete" is refused exactly when you -need to prune. Listing needs `pull`, so this holds both tokens. - -- keeps: every tag named by a receipt in receipts/*.json (plus 'latest') -- targets: all other stream tags, and every manifest in stream/cache -- auth: ATCR_APP_PASSWORD from env (--keychain for the macOS helper) +"""Prune atcr.io/zat.dev/stream via the PDS records atcr indexes. + +atcr has no registry-API manifest DELETE (every DELETE returns 405, even +correctly scoped — recorded 2026-08-09). Images are really deleted by +removing their `io.atcr.tag` / `io.atcr.manifest` records from our own PDS, +which is what atcr indexes. That path is not quota-blocked. atcr's +server-side GC recomputes quota roughly daily, so usage lags deletion by up +to ~24h. + +plan: +- inventory: public listRecords over io.atcr.tag + io.atcr.manifest +- scope: ONLY repository == "stream" or "stream/cache" — the same + collections hold other projects' images (zds), which must never be touched +- keep: every tag named by a receipt in receipts/*.json, plus 'latest'; + manifests stay if reachable from a kept tag (index children included, + resolved through the raw manifest blob) +- delete: everything else stream-scoped, via com.atproto.repo.applyWrites + +auth: ATCR_USER (default zat.dev) + ATCR_APP_PASSWORD, an atproto app +password — createSession against the account's PDS. Reads are public. usage: scripts/registry-prune [--dry-run] """ import json +import os import pathlib -import subprocess import sys -import urllib.request import urllib.error +import urllib.request -REGISTRY = "https://atcr.io" -REPOS = ["zat.dev/stream", "zat.dev/stream/cache"] +HANDLE = os.environ.get("ATCR_USER", "zat.dev") +REPOSITORIES = {"stream", "stream/cache"} DRY = "--dry-run" in sys.argv +BATCH = 200 # applyWrites cap + + +def get_json(url: str) -> dict: + with urllib.request.urlopen(url, timeout=30) as resp: + return json.load(resp) -def credential(): - """ATCR_USER/ATCR_APP_PASSWORD env (CI: the pipeline secret) first; - docker-credential-osxkeychain only as an explicit opt-in fallback.""" - import os - if os.environ.get("ATCR_APP_PASSWORD"): - return os.environ.get("ATCR_USER", "zat.dev"), os.environ["ATCR_APP_PASSWORD"] - if "--keychain" not in sys.argv: - sys.exit("no ATCR_APP_PASSWORD in env; pass --keychain to use the macOS helper (prompts!)") - out = subprocess.run( - ["docker-credential-osxkeychain", "get"], - input="https://atcr.io", capture_output=True, text=True, +def resolve_pds(handle: str) -> tuple[str, str]: + did = get_json( + "https://public.api.bsky.app/xrpc/com.atproto.identity.resolveHandle" + f"?handle={handle}" + )["did"] + doc = get_json(f"https://plc.directory/{did}") + pds = next( + s["serviceEndpoint"] for s in doc["service"] if s["id"] == "#atproto_pds" ) - if out.returncode != 0 or not out.stdout.strip(): - sys.exit(f"keychain refused: {out.stderr.strip() or out.stdout.strip()}") - d = json.loads(out.stdout) - return d["Username"], d["Secret"] - - -def bearer(user, secret, repo, actions): - """Token for one scope. Ask for exactly the actions you need. - - `delete` must be requested ALONE: atcr refuses pull/push while an account - is over quota, so a combined "pull,push,delete" is denied precisely when - pruning is the thing you need to do (upstream confirmed the server assumed - clients would ask for delete on its own). Listing still needs `pull`, so - callers hold both tokens. - """ - try: - urllib.request.urlopen(f"{REGISTRY}/v2/{repo}/tags/list") - except urllib.error.HTTPError as e: - challenge = e.headers.get("WWW-Authenticate", "") - if not challenge.startswith("Bearer "): - return None # registry accepts basic auth directly - parts = dict( - kv.split("=", 1) for kv in challenge[len("Bearer "):].split(",") + return did, pds + + +def list_records(pds: str, did: str, collection: str) -> list[dict]: + records, cursor = [], None + while True: + url = ( + f"{pds}/xrpc/com.atproto.repo.listRecords" + f"?repo={did}&collection={collection}&limit=100" ) - realm = parts["realm"].strip('"') - service = parts.get("service", "").strip('"') - import base64 - basic = "Basic " + base64.b64encode(f"{user}:{secret}".encode()).decode() - from urllib.parse import quote - scope = quote(f"repository:{repo}:{actions}", safe="") - url = f"{realm}?service={service}&scope={scope}&account={quote(user, safe='')}" - req = urllib.request.Request(url) - req.add_header("Authorization", basic) - try: - body = json.load(urllib.request.urlopen(req)) - except urllib.error.HTTPError as e: - print(f" token scope '{actions}' -> {e.code}: {e.read()[:120].decode(errors='replace')}") - return None - return body.get("token") or body.get("access_token") - return None - - -def api(method, path, token, user, secret, headers=None): - req = urllib.request.Request(REGISTRY + path, method=method) - if token: - req.add_header("Authorization", f"Bearer {token}") - else: - import base64 - req.add_header("Authorization", "Basic " + base64.b64encode(f"{user}:{secret}".encode()).decode()) - for k, v in (headers or {}).items(): - req.add_header(k, v) - return urllib.request.urlopen(req) - - -def main(): - keep = {"latest"} - for r in pathlib.Path("receipts").glob("*.json"): - keep.add(r.stem) - print(f"keeping tags: {sorted(keep)}") - - user, secret = credential() - accept = {"Accept": "application/vnd.docker.distribution.manifest.v2+json, application/vnd.oci.image.manifest.v1+json"} - - for repo in REPOS: - token = bearer(user, secret, repo, "pull") - del_token = bearer(user, secret, repo, "delete") - print(f"{repo}: pull={'ok' if token else 'DENIED'} delete={'ok' if del_token else 'DENIED'}") - try: - with api("GET", f"/v2/{repo}/tags/list?n=1000", token, user, secret) as resp: - tags = json.load(resp).get("tags") or [] - except urllib.error.HTTPError as e: - print(f"{repo}: tags/list -> {e.code} (skipping)") + if cursor: + url += f"&cursor={cursor}" + page = get_json(url) + records += page["records"] + cursor = page.get("cursor") + if not cursor or not page["records"]: + return records + + +def rkey(uri: str) -> str: + return uri.rsplit("/", 1)[1] + + +def bare_digest(d: str) -> str: + return d.removeprefix("sha256:") + + +def index_children(pds: str, did: str, record: dict) -> set[str]: + """Child manifest digests of an OCI index, from the raw manifest blob.""" + blob = record["value"].get("manifestBlob") + if not blob: + return set() + cid = blob["ref"]["$link"] + with urllib.request.urlopen( + f"{pds}/xrpc/com.atproto.sync.getBlob?did={did}&cid={cid}", timeout=30 + ) as resp: + manifest = json.load(resp) + return {bare_digest(m["digest"]) for m in manifest.get("manifests", [])} + + +def create_session(pds: str) -> dict: + password = os.environ.get("ATCR_APP_PASSWORD") or sys.exit( + "no ATCR_APP_PASSWORD in env" + ) + req = urllib.request.Request( + f"{pds}/xrpc/com.atproto.server.createSession", + data=json.dumps({"identifier": HANDLE, "password": password}).encode(), + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=30) as resp: + return json.load(resp) + + +def apply_deletes(pds: str, session: dict, doomed: list[tuple[str, str]]) -> None: + for start in range(0, len(doomed), BATCH): + writes = [ + { + "$type": "com.atproto.repo.applyWrites#delete", + "collection": collection, + "rkey": key, + } + for collection, key in doomed[start : start + BATCH] + ] + req = urllib.request.Request( + f"{pds}/xrpc/com.atproto.repo.applyWrites", + data=json.dumps({"repo": session["did"], "writes": writes}).encode(), + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {session['accessJwt']}", + }, + ) + urllib.request.urlopen(req, timeout=60).close() + print(f" deleted {len(writes)} records") + + +def main() -> None: + keep_tags = {"latest"} + for receipt in pathlib.Path("receipts").glob("*.json"): + keep_tags.add(receipt.stem) + print(f"keeping tags: {sorted(keep_tags)}") + + did, pds = resolve_pds(HANDLE) + print(f"{HANDLE} = {did} @ {pds}") + + tags = [ + t + for t in list_records(pds, did, "io.atcr.tag") + if t["value"].get("repository") in REPOSITORIES + ] + manifests = { + rkey(m["uri"]): m + for m in list_records(pds, did, "io.atcr.manifest") + if m["value"].get("repository") in REPOSITORIES + } + print(f"stream-scoped: {len(tags)} tags, {len(manifests)} manifests") + + # closure of manifests reachable from kept tags (cache tags never kept) + keep_digests: set[str] = set() + doomed_tags: list[dict] = [] + for tag in tags: + value = tag["value"] + kept = value["repository"] == "stream" and value["tag"] in keep_tags + if not kept: + doomed_tags.append(tag) + continue + digest = rkey(value["manifest"]) + keep_digests.add(digest) + if digest in manifests: + keep_digests |= index_children(pds, did, manifests[digest]) + + doomed: list[tuple[str, str]] = [] + for tag in doomed_tags: + print(f" doomed tag {rkey(tag['uri'])} ({tag['value']['tag']})") + doomed.append(("io.atcr.tag", rkey(tag["uri"]))) + for digest, record in sorted(manifests.items()): + if digest in keep_digests: continue - print(f"{repo}: {len(tags)} tags") - for tag in tags: - doomed = repo.endswith("/cache") or tag not in keep - if not doomed: - print(f" keep {tag}") - continue - try: - with api("HEAD", f"/v2/{repo}/manifests/{tag}", token, user, secret, accept) as resp: - digest = resp.headers["Docker-Content-Digest"] - except urllib.error.HTTPError as e: - print(f" ? {tag}: HEAD -> {e.code}") - continue - if DRY: - print(f" would-delete {tag} ({digest[:19]}…)") - continue - try: - api("DELETE", f"/v2/{repo}/manifests/{digest}", del_token, user, secret).close() - print(f" DELETED {tag} ({digest[:19]}…)") - except urllib.error.HTTPError as e: - print(f" FAILED {tag}: DELETE -> {e.code} {e.read()[:120].decode(errors='replace')}") - - print("done. registry GC timing is server-side; quota may lag deletion.") + print(f" doomed manifest {digest[:19]}… ({record['value']['repository']})") + doomed.append(("io.atcr.manifest", digest)) + + kept = len(tags) - len(doomed_tags) + print(f"kept {kept} tags / {len(keep_digests)} reachable manifests; " + f"{len(doomed)} records doomed") + if not doomed: + print("nothing to prune.") + return + if DRY: + print("dry run: no deletes issued.") + return + apply_deletes(pds, create_session(pds), doomed) + print("done. atcr's server-side GC recomputes quota within ~24h.") if __name__ == "__main__": -- 2.51.2