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__":