Library media from the index run folder (<corpus>/<run>/library-media/media-<stamp>.tar, newest wins, else the latest earlier run's) — indexing PR #24 (ace1e6b), via this image's aws-CLI artifact_store (GPU variant)

pipeline/library_media.py

@@ -1,47 +1,133 @@
 """Library media for the Space's Explorer: the joints viewer's models, moving-parts and drop-test clips.
 
 SimReady Central builds these per asset — model.glb + articulation.json for each prop with moving joints
-(its scripts/library_media/build.py), the moving-parts clips its lab render pool renders, and the drop tests — and
-they are kept in the artifact bucket next to the index runs, one archive per corpus:
-``<ARTIFACT_S3_URI>/library-media/<CORPUS_NAME>/media.tar``, an uncompressed tar of ``manifest.json`` +
-``<id>/<file>`` (its members are already compressed). Outside ``<corpus>/``, so nothing that lists a corpus's runs
-sees it; corpus-level, not per run, because the media are keyed by the assets' URLs, not by an index build.
-
-Run once at container start by supervisord (program ``library-media``), this downloads that ONE object with the
-same aws CLI and credentials as the index restore (artifact_store), unpacks it and installs the tree as
-``.build/library_media/<INDEX_KEY>/``; nginx serves it at /library_media/ (docker/hf-space/nginx.conf.template), where
-the Explorer asks for /library_media/usdsearch-dev/manifest.json — usdsearch-dev being its index key in /config.js.
-NOT under ``.build/restore/``: restore.download empties that whole directory at every start.
-
-Nothing here is needed to search. Any failure — no archive for this corpus, no bucket credentials, a network error,
-an unsafe or malformed archive — leaves the previous tree (none at boot): nginx answers 404 and the Explorer shows no
-media. LIBRARY_MEDIA=off disables the fetch.
+(its scripts/library_media/build.py), the moving-parts clips its lab render pool renders, and the drop tests. They
+live with the index run they belong to, next to its hierarchy/ (pipeline/hierarchy/store.py):
+
+    <ARTIFACT_S3_URI>/<corpus>/<run_id>/library-media/media-<YYYYMMDDTHHMMZ>.tar
+
+an uncompressed tar of ``manifest.json`` + ``<id>/<file>`` (its members are already compressed). Each upload is a NEW
+file, named by when it was made, never written over: the newest one in the run is served, so a bad upload is
+reverted by deleting it (the one before it comes back), and pinning the Space to an earlier run (RESTORE_RUN_ID)
+brings back that run's media with its index.
+
+Run once at container start by supervisord (program ``library-media``), this finds the run the Space restores (the
+same rule as hf_space_seed.resolve_prefix: RESTORE_RUN_ID, else the corpus's latest.json), takes that run's newest
+archive — else the newest one of the latest EARLIER run that has any (a new index run keeps the media until its own
+are uploaded; a later run's never apply to an earlier one) — downloads it with the same aws CLI and credentials as
+the index restore (artifact_store; CliStore below), unpacks it and installs the tree as ``.build/library_media/<INDEX_KEY>/``; nginx
+serves it at /library_media/ (docker/hf-space/nginx.conf.template), where the Explorer asks for
+/library_media/usdsearch-dev/manifest.json — usdsearch-dev being its index key in /config.js. NOT under
+``.build/restore/``: restore.download empties that whole directory at every start.
+
+Nothing here is needed to search. Any failure — no archive, no bucket credentials, a network error, an unsafe or
+malformed archive — leaves the previous tree (none at boot): nginx answers 404 and the Explorer shows no media.
+LIBRARY_MEDIA=off disables the fetch.
 """
 
 from __future__ import annotations
 
 import json
 import os
+import re
 import shutil
 import subprocess
 import sys
 import tarfile
 import time
 from pathlib import Path
+from typing import Optional
 
 from .config import BUILD_DIR, Config
 
-ARCHIVE = "media.tar"
-PREFIX = "library-media"
+SUBDIR = "library-media"
+ARCHIVE = re.compile(r"^media-\d{8}T\d{4}Z\.tar$")   # media-20261005T1010Z.tar: the name sorts by time
 # The Explorer's index key for this Space (/config.js in nginx.conf.template): the manifest's URL names it.
 INDEX_KEY = "usdsearch-dev"
 MEDIA_ROOT = BUILD_DIR / "library_media"
 DISABLED = {"off", "none", "false", "0"}
 
 
-def archive_uri(cfg: Config, corpus: str) -> str:
-    """Where this corpus's media archive lives in the artifact bucket."""
-    return f"{cfg.require('ARTIFACT_S3_URI').rstrip('/')}/{PREFIX}/{corpus}/{ARCHIVE}"
+class CliStore:
+    """What fetch() reads from the bucket, over this image's aws-CLI artifact_store (before indexing PR #25, which
+    adds read_text / list_objects / list_subprefixes / download_file in boto3)."""
+
+    def __init__(self):
+        from . import artifact_store
+        self.s = artifact_store
+
+    def bucket_base(self, cfg: Config) -> str:
+        return self.s.bucket_base(cfg)
+
+    def _run(self, cfg: Config, args: list, timeout: int):
+        return subprocess.run(self.s.aws_cmd(cfg) + args, capture_output=True, text=True, env=self.s.artifact_env(),
+                              timeout=timeout)
+
+    @staticmethod
+    def _missing(r) -> bool:
+        err = r.stderr or r.stdout or ""
+        return "(404)" in err or "Not Found" in err or "NoSuchKey" in err or "does not exist" in err
+
+    def read_text(self, cfg: Config, uri: str) -> Optional[str]:
+        r = self._run(cfg, ["cp", uri, "-"], 120)
+        if r.returncode == 0:
+            return r.stdout
+        if self._missing(r):
+            return None
+        raise RuntimeError(f"aws s3 cp {uri} - failed (exit {r.returncode}): {(r.stderr or '').strip()[:300]}")
+
+    def _ls(self, cfg: Config, uri: str) -> list:
+        r = self._run(cfg, ["ls", uri.rstrip("/") + "/"], 120)
+        if r.returncode == 0:
+            return r.stdout.splitlines()
+        if r.returncode == 1 and not (r.stderr or "").strip():   # aws s3 ls: nothing under the prefix
+            return []
+        raise RuntimeError(f"aws s3 ls {uri} failed (exit {r.returncode}): {(r.stderr or '').strip()[:300]}")
+
+    def list_subprefixes(self, cfg: Config, uri: str) -> list:
+        return [ln.split()[-1].rstrip("/") for ln in self._ls(cfg, uri) if ln.strip().startswith("PRE ")]
+
+    def list_objects(self, cfg: Config, uri: str) -> list:
+        rows = [ln.split(None, 3) for ln in self._ls(cfg, uri) if not ln.strip().startswith("PRE ")]
+        return [(row[3], int(row[2])) for row in rows if len(row) == 4 and row[2].isdigit()]
+
+    def download_file(self, cfg: Config, uri: str, dest: Path, attempts: int = 3) -> bool:
+        dest.parent.mkdir(parents=True, exist_ok=True)
+        for attempt in range(1, attempts + 1):
+            r = self._run(cfg, ["cp", uri, str(dest), "--only-show-errors"], 1800)
+            if r.returncode == 0:
+                return True
+            if self._missing(r):
+                return False
+            print(f"==> library media: aws s3 cp failed (attempt {attempt}/{attempts}, exit {r.returncode}): "
+                  f"{(r.stderr or r.stdout).strip()[:300]}")
+            if attempt < attempts:
+                time.sleep(5 * attempt)
+        raise RuntimeError(f"aws s3 cp {uri} failed {attempts} times")
+
+
+def run_prefix(cfg: Config, store, corpus: str) -> str:
+    """The run the Space restores: RESTORE_RUN_ID, else latest.json (hf_space_seed.resolve_prefix's rule)."""
+    base = f"{store.bucket_base(cfg)}/{corpus}"
+    pinned = cfg.get("RESTORE_RUN_ID").strip()
+    if pinned:
+        return f"{base}/{pinned}"
+    pointer = store.read_text(cfg, f"{base}/latest.json")
+    if pointer is None:
+        raise SystemExit(f"no published run at {base}/latest.json")
+    return json.loads(pointer)["prefix"].rstrip("/")
+
+
+def archive_uri(cfg: Config, store, prefix: str) -> Optional[str]:
+    """The newest archive of the run at ``prefix``, else of the latest earlier run that has any; None: there is none.
+    Run ids start with the date (hierarchy/store.py), so they sort by age."""
+    base, run = prefix.rsplit("/", 1)
+    earlier = [r for r in sorted(store.list_subprefixes(cfg, base), reverse=True) if r < run]
+    for r in [run] + earlier:
+        names = [rel for rel, _ in store.list_objects(cfg, f"{base}/{r}/{SUBDIR}") if ARCHIVE.match(rel)]
+        if names:
+            return f"{base}/{r}/{SUBDIR}/{max(names)}"
+    return None
 
 
 def valid_manifest(path: Path) -> int | None:
@@ -79,27 +165,9 @@ def install(src: Path, dest: Path) -> None:
     shutil.rmtree(old, ignore_errors=True)
 
 
-def cli_download(cfg: Config, uri: str, dest: Path, attempts: int = 3) -> bool:
-    """``aws s3 cp`` ``uri`` to ``dest``; False when there is no such object (this image's artifact_store is the
-    aws-CLI one, before indexing PR #25's ``download_file``)."""
-    from .artifact_store import artifact_env, aws_cmd
-    dest.parent.mkdir(parents=True, exist_ok=True)
-    for attempt in range(1, attempts + 1):
-        r = subprocess.run(aws_cmd(cfg) + ["cp", uri, str(dest), "--only-show-errors"],
-                           capture_output=True, text=True, env=artifact_env(), timeout=1800)
-        if r.returncode == 0:
-            return True
-        err = (r.stderr or r.stdout).strip()
-        if "(404)" in err or "Not Found" in err or "does not exist" in err:
-            return False
-        print(f"==> library media: aws s3 cp failed (attempt {attempt}/{attempts}, exit {r.returncode}): {err[:300]}")
-        if attempt < attempts:
-            time.sleep(5 * attempt)
-    raise RuntimeError(f"aws s3 cp {uri} failed {attempts} times")
-
-
-def fetch(env: dict[str, str], root: Path = MEDIA_ROOT, download=None) -> int:
-    """Fetch this corpus's media into ``root/INDEX_KEY``; returns the number of assets installed (0: none)."""
+def fetch(env: dict[str, str], root: Path = MEDIA_ROOT, store=None) -> int:
+    """Fetch this corpus's media into ``root/INDEX_KEY``; returns the number of assets installed (0: none).
+    ``store``: artifact_store's read_text / list_subprefixes / list_objects / download_file / bucket_base."""
     corpus = env.get("CORPUS_NAME", "").strip()
     if env.get("LIBRARY_MEDIA", "").strip().lower() in DISABLED:
         print("==> library media: off (LIBRARY_MEDIA)")
@@ -107,18 +175,22 @@ def fetch(env: dict[str, str], root: Path = MEDIA_ROOT, download=None) -> int:
     if not corpus or "/" in corpus or corpus.startswith("."):
         print(f"==> library media: no usable CORPUS_NAME ({corpus!r}), skipped")
         return 0
-    if download is None:
-        download = cli_download
+    if store is None:
+        store = CliStore()
     cfg = Config(values=dict(env))
-    uri = archive_uri(cfg, corpus)
+    t0 = time.time()
+    prefix = run_prefix(cfg, store, corpus)
+    uri = archive_uri(cfg, store, prefix)
+    if uri is None:
+        print(f"==> library media: none for {prefix} or an earlier run — none served")
+        return 0
     root.mkdir(parents=True, exist_ok=True)
     tmp = root / f".{INDEX_KEY}.download"
     shutil.rmtree(tmp, ignore_errors=True)
-    t0 = time.time()
     try:
-        archive = tmp / ARCHIVE
-        if not download(cfg, uri, archive):
-            print(f"==> library media: no {uri} — none served")
+        archive = tmp / "media.tar"
+        if not store.download_file(cfg, uri, archive):
+            print(f"==> library media: {uri} is gone — none served")
             return 0
         unpack(archive, tmp / "tree")
         count = valid_manifest(tmp / "tree" / "manifest.json")