feat(depot): share catalog versions through remotes

CHANGELOG.md

@@ -8,6 +8,7 @@ schemas in `nnnotes.schemas` and the documented Python API. Before 1.0 a minor v
 
 ### Features
 
+- Share catalog versions and labels through depot remotes, with verified catalog files fetched on demand.
 - Store content, task inputs, results and catalog versions in a persistent depot with optional remote storage, verified collection and bounded source downloads.
 - Release recoverable bundle downloads after successful conversion batches under the configured budget and command leases.
 - Assemble Live2D, room, spot, live, story and UI products from shared object tables and explicit manifest references.

docs/contracts.md

@@ -295,17 +295,24 @@ ac/ab/<key>.json                results
 inputs/<kind>/ab/<digest>.json  input identities by name
 runs/<run>.json                 run manifests
 runs/selections/<digest>        the id of the latest run of a selection
+catalogs/versions/<id>.json      one catalog version and its labels
+catalogs/<sha256>.bin           catalog files
 ```
 
 The depot directory is the remote's cache. A content, result, input identity or run the depot lacks is read from the
 remote and kept; a content read from the remote is checked against its content id. What the depot adds goes to the
 remote as well: the bytes a result names before the result, a run manifest before the pointer to it. A result that
 differs from the remote's result of the same key is a conflict, as in the depot itself. The latest run of a selection
 is the one any machine wrote last, and the results of its tasks are known to be on the remote without asking for each
-of them. Downloads, file identities by path (`inputs/paths/`), catalog versions, measured costs, run logs and leases
+of them. Downloads, file identities by path (`inputs/paths/`), measured costs, run logs and leases
 stay in the depot directory.
 
-Every request names one object; only `depot gc --remote` lists the remote.
+Catalog discovery lists the version records and merges them into the local catalog index. Catalog bytes are fetched
+on demand and verified by size and SHA-256 before use. Each import publishes its catalog files before its version
+record. `--stored-catalog` can select a version held by the remote on a fresh depot. Import sequences order the
+versions, with version ids breaking ties between independent imports.
+
+Object reads and writes name individual keys. Catalog discovery and `depot gc --remote` use listings.
 
 ## UI contracts
 

src/nnnotes/cli/assets.py

@@ -333,12 +333,13 @@ def __setstate__(self, state):
     def catalog(self, vid: str):
         """(Catalog, {location id: location}) of a stored catalog version."""
         from ..pipeline import catalogdb
+        from ..pipeline.depot import open_depot
         from ..source.catalog import Catalog
 
         with self._lock:
             hit = self._catalogs.get(vid)
             if hit is None:
-                db = catalogdb.CatalogDB(self.root)
+                db = catalogdb.CatalogDB(self.root, remote=open_depot(self.cfg, root=self.root).remote)
                 v = next((x for x in db.versions() if x["id"] == vid), None)
                 if v is None:
                     raise FileNotFoundError(f"catalog version {vid[:12]} is not in the depot")
@@ -524,7 +525,7 @@ def __init__(self, args, cfg, log=_log, layouts=()):
         self.apk = cfg.path("paths", "apk")
         self.fetcher = CatalogFetcher(self.root, cfg)
         self.depot = open_depot(cfg, fetch=self.fetcher)
-        self.db = CatalogDB(self.root)
+        self.db = CatalogDB(self.root, remote=self.depot.remote)
         try:
             self.selection = parse_selection(getattr(args, "select", None))
         except ValueError as e:

src/nnnotes/cli/catalog.py

@@ -46,8 +46,10 @@ def cmd_browse(args, cfg):
 # ---------------------------------------------------------------- catalog versions of the depot
 def _db(cfg):
     from ..pipeline.catalogdb import CatalogDB
+    from ..pipeline.depot import open_depot
 
-    return CatalogDB(Session(cfg).depot_root())
+    depot = open_depot(cfg)
+    return CatalogDB(depot.root, remote=depot.remote)
 
 
 def version_line(v: dict) -> str:

src/nnnotes/pipeline/catalogdb.py

@@ -16,6 +16,10 @@
                                          catalogs (sha256, size, buildResultHash), the CDN's `.hash` text when it was
                                          served (an opaque change signal), resource version, APK version name
 
+With a depot remote, each version has its own `catalogs/versions/<id>.json` object. Version discovery merges these
+records into the local index; catalog bytes are fetched and verified only when read. Import sequences order the
+versions, with version ids breaking ties between independent imports.
+
 `diff` compares two indexes: keys (the primary keys of asset locations: internal id, provider, resource type and
 dependencies, bundles and raw files named by their stable names), bundles and raw files matched by stable name (the
 name without its content hash: added, removed, changed with old and new name, size, hash, CRC) and the keys whose
@@ -36,6 +40,7 @@
 from . import contract
 from .contract import Cost, Input
 from .depot import write_file
+from .remote import Remote
 from .stages import Output, Stage
 
 CATALOGS = "nnnotes.catalogs/1"
@@ -413,8 +418,9 @@ def fetch(cdn: str, language: str, timeout: float = 120, *, version: str = "main
 class CatalogDB:
     """The catalog versions of a depot (<depot>/catalogs/)."""
 
-    def __init__(self, depot_root):
+    def __init__(self, depot_root, *, remote: Remote | None = None):
         self.root = Path(depot_root) / "catalogs"
+        self.remote = remote
 
     def path(self, sha: str) -> Path:
         if not contract.is_sha256(sha):
@@ -425,10 +431,32 @@ def _load(self) -> list[dict]:
         try:
             doc = contract.loads((self.root / "index.json").read_bytes())
         except FileNotFoundError:
-            return []
+            doc = {"schema": CATALOGS, "versions": []}
         if doc.get("schema") != CATALOGS:
             raise ValueError(f"{self.root / 'index.json'} is not a {CATALOGS} document")
-        return doc["versions"]
+        versions = {v["id"]: v for v in doc["versions"]}
+        if self.remote is not None:
+            for key, _, _ in self.remote.list("catalogs/versions/"):
+                if not key.endswith(".json"):
+                    continue
+                data = self.remote.get(key)
+                if data is None:
+                    continue
+                v = contract.loads(data)
+                expected = version_id(v["remote"]["sha256"], v["apk"]["sha256"] if v["apk"] else None, v.get("source"))
+                if v["id"] != expected or key != f"catalogs/versions/{expected}.json":
+                    raise ValueError(f"catalog version identity differs: {key}")
+                local = versions.get(expected)
+                if local is not None:
+                    v["labels"] = sorted(set(v["labels"]) | set(local["labels"]))
+                    for name, value in local.items():
+                        if v.get(name) is None:
+                            v[name] = value
+                versions[expected] = v
+        merged = sorted(versions.values(), key=lambda v: (v["seq"], v["id"]))
+        if self.remote is not None:
+            self._save(merged)
+        return merged
 
     def _save(self, versions: list[dict]) -> None:
         write_file(self.root / "index.json", contract.encode({"schema": CATALOGS, "versions": versions}))
@@ -444,8 +472,12 @@ def versions(self, region: str | None = None, language: str | None = None) -> li
     def _put(self, data: bytes) -> dict:
         sha = contract.sha256(data)
         p = self.path(sha)
-        if not (p.is_file() and p.stat().st_size == len(data)):
+        if not (p.is_file() and p.read_bytes() == data):
             write_file(p, data)
+        if self.remote is not None:
+            key = f"catalogs/{sha}.bin"
+            if self.remote.size(key) != len(data):
+                self.remote.put(key, data)
         return {"sha256": sha, "size": len(data), "buildResultHash": parse_header(data)["buildResultHash"]}
 
     def add(
@@ -507,6 +539,8 @@ def add(
                 raise ValueError(f"label {label!r} already names catalog version {other['id'][:12]}")
         if label not in v["labels"]:
             v["labels"].append(label)
+        if self.remote is not None:
+            self.remote.put(f"catalogs/versions/{vid}.json", contract.encode(v))
         self._save(versions)
         return v
 
@@ -532,10 +566,26 @@ def get(self, ref: str, region: str | None = None, language: str | None = None)
 
     def catalog_bytes(self, version: dict) -> tuple[bytes, bytes | None]:
         """(remote catalog, APK catalog or None) of a version."""
-        remote = self.path(version["remote"]["sha256"]).read_bytes()
-        apk = self.path(version["apk"]["sha256"]).read_bytes() if version["apk"] else None
+        remote = self._bytes(version["remote"])
+        apk = self._bytes(version["apk"]) if version["apk"] else None
         return remote, apk
 
+    def _bytes(self, record: dict) -> bytes:
+        sha, size = record["sha256"], record["size"]
+        path = self.path(sha)
+        try:
+            data = path.read_bytes()
+        except FileNotFoundError:
+            data = self.remote.get(f"catalogs/{sha}.bin") if self.remote is not None else None
+            if data is None:
+                raise FileNotFoundError(f"catalog content {sha} is not in the depot") from None
+            if len(data) != size or contract.sha256(data) != sha:
+                raise ValueError(f"catalog content {sha} differs from its identity") from None
+            write_file(path, data)
+        if len(data) != size or contract.sha256(data) != sha:
+            raise ValueError(f"catalog content {sha} differs from its identity")
+        return data
+
     def index(self, version: dict) -> dict:
         return index(*self.catalog_bytes(version))
 
@@ -545,6 +595,7 @@ def inputs(self, version: dict) -> list[Input]:
         for role in CATALOG_NAMES:
             rec = version[role]
             if rec:
+                self._bytes(rec)
                 out.append(
                     Input(
                         role,

src/nnnotes/pipeline/depot.py

@@ -725,13 +725,14 @@ def _gc_remote(self, remote: Remote, keep: int | None, dry_run: bool, grace: flo
         return out
 
 
-def open_depot(cfg, *, fetch=None) -> Depot:
+def open_depot(cfg, *, fetch=None, root=None) -> Depot:
     """The depot of a configuration: `[paths] depot`, with the remote `[depot] remote` names (its `endpoint`,
-    `access_key_id`, `secret_access_key` and `signing_region`) when it is set."""
+    `access_key_id`, `secret_access_key` and `signing_region`) when it is set. `root` selects an explicit depot
+    directory when the caller already holds one."""
     from ..source.config import ConfigError
     from ..source.session import Session
 
-    root = Session(cfg).depot_root()
+    root = Session(cfg).depot_root() if root is None else Path(root)
     url = cfg.get("depot", "remote")
     if not url:
         return Depot(root, fetch=fetch)

src/nnnotes/pipeline/remote.py

@@ -7,12 +7,14 @@
                                              sha256 of the name)
     <remote>/runs/<run>.json                 run manifests
     <remote>/runs/selections/<digest>        the id of the latest run of a selection
+    <remote>/catalogs/versions/<id>.json     one catalog version and its labels
+    <remote>/catalogs/<sha>.bin              catalog bytes, verified when fetched
 
 A remote is `s3://<bucket>/<prefix>` (S3-compatible object storage: path-style requests, checksums only where the
 request needs them, so stores other than AWS answer them) or `file://<directory>` (a shared directory). Every
 operation but `list` names one object, so a remote behind a gateway that caches listings answers them all; `list` is
-used by `depot gc` only. Contents and results are written once: a write is skipped when the object is there with the
-same size. The local depot is the remote's cache (pipeline.depot).
+used for catalog version discovery and `depot gc`. Contents and results are written once: a write is skipped when
+the object is there with the same size. The local depot is the remote's cache (pipeline.depot).
 """
 
 from __future__ import annotations

tests/cli/test_assets.py

@@ -485,6 +485,34 @@ def test_a_second_depot_with_the_same_remote_hits_without_downloads(tmp_path, ca
     assert not [p for p in (tmp_path / "b").glob("downloads/**/*") if p.is_file()]  # nothing downloaded
 
 
+def test_a_fresh_depot_exports_a_stored_catalog_from_its_remote(tmp_path, capsys, fake, monkeypatch):
+    monkeypatch.setenv("NNNOTES_DEPOT_REMOTE", (tmp_path / "remote").as_uri())
+    d = setup_data(tmp_path, depots=("a",))
+    code, out, err = export(capsys, tmp_path, d, "a", "out-a")
+    assert code == 1, err
+    first = json.loads(out)
+    code, out, err = nn(
+        capsys,
+        "export",
+        "assets",
+        "--stored-catalog",
+        "latest",
+        "--depot",
+        tmp_path / "b",
+        "-o",
+        tmp_path / "out-b",
+        "--workers",
+        "0",
+        "--json",
+    )
+    assert code == 1 and out, err
+    again = json.loads(out)
+    assert again["tasks"]["ran"] == 0 and again["run"] == first["run"], err
+    assert tree(tmp_path / "out-a", [f"{assets.REPORTS}/run.json"]) == tree(
+        tmp_path / "out-b", [f"{assets.REPORTS}/run.json"]
+    )
+
+
 def test_export_is_the_same_with_worker_processes(tmp_path, capsys, fake):
     d = setup_data(tmp_path, depots=("s0", "s2"))
     export(capsys, tmp_path, d, "s0", "o0")

tests/pipeline/test_catalog_remote.py

@@ -0,0 +1,85 @@
+"""Catalog metadata and content shared by independent depots over a directory remote."""
+
+import pytest
+
+from nnnotes.pipeline import contract
+from nnnotes.pipeline.catalogdb import CatalogDB
+from nnnotes.pipeline.remote import open_remote
+from tests.pipeline.test_catalogdb import APK, H3, H4, build, remote_entries
+
+
+def databases(tmp_path):
+    remote = open_remote((tmp_path / "remote").as_uri())
+    return remote, CatalogDB(tmp_path / "a", remote=remote), CatalogDB(tmp_path / "b", remote=remote)
+
+
+def test_catalog_versions_are_shared_and_content_is_fetched_on_demand(tmp_path):
+    remote, first, second = databases(tmp_path)
+    body, apk = build(remote_entries()), build(APK)
+    version = first.add(body, apk, label="stable", region="one", language="en")
+    assert remote.get(f"catalogs/versions/{version['id']}.json") == contract.encode(version)
+    assert remote.get("catalogs/index.json") is None
+    assert second.get("stable") == version
+    assert not second.path(version["remote"]["sha256"]).exists()
+    assert second.catalog_bytes(version) == (body, apk)
+    assert second.path(version["apk"]["sha256"]).read_bytes() == apk
+    assert CatalogDB(tmp_path / "b").get("stable") == version
+    assert CatalogDB(tmp_path / "b").catalog_bytes(version) == (body, apk)
+
+
+def test_independent_imports_preserve_versions_and_refresh_known_metadata(tmp_path):
+    _, first, second = databases(tmp_path)
+    body = build(remote_entries())
+    one = first.add(body, label="one", region="r", language="en")
+    assert second.get("one")["id"] == one["id"]
+    two = second.add(build(remote_entries(h_a=H3)), label="two", region="r", language="en")
+    three = first.add(build(remote_entries(h_a=H4)), label="three", region="r", language="en")
+    assert [v["id"] for v in second.versions()] == [one["id"], two["id"], three["id"]]
+    first.add(body, label="extra", region="r", language="en", resource_version="1.2.3")
+    assert second.get("extra")["resourceVersion"] == "1.2.3"
+    assert second.get("latest")["id"] == three["id"]
+    with pytest.raises(ValueError, match="already names"):
+        second.add(build(remote_entries(h_a=H3)), label="one", region="r", language="en")
+
+
+def test_remote_catalog_bytes_are_verified_before_local_storage(tmp_path):
+    remote, first, second = databases(tmp_path)
+    body = build(remote_entries())
+    version = first.add(body)
+    sha = version["remote"]["sha256"]
+    remote.put(f"catalogs/{sha}.bin", b"x" * len(body))
+    with pytest.raises(ValueError, match="differs from its identity"):
+        second.catalog_bytes(version)
+    assert not second.path(sha).exists()
+
+
+def test_remote_catalog_record_must_match_its_identity(tmp_path):
+    remote, first, second = databases(tmp_path)
+    version = first.add(build(remote_entries()))
+    version["id"] = "0" * 64
+    remote.put(f"catalogs/versions/{version['id']}.json", contract.encode(version))
+    with pytest.raises(ValueError, match="version identity differs"):
+        second.versions()
+
+
+def test_catalog_task_inputs_materialize_shared_catalogs(tmp_path):
+    _, first, second = databases(tmp_path)
+    body, apk = build(remote_entries()), build(APK)
+    version = first.add(body, apk)
+    inputs = second.inputs(second.get(version["id"]))
+    assert [i.sha256 for i in inputs] == [contract.sha256(body), contract.sha256(apk)]
+    assert second.catalog_bytes(version) == (body, apk)
+
+
+def test_catalog_locator_recovery_reads_shared_version_metadata(tmp_path):
+    from nnnotes.cli.assets import CatalogFetcher
+    from nnnotes.source.config import Config
+
+    remote, first, _ = databases(tmp_path)
+    body = build(remote_entries())
+    version = first.add(body)
+    cfg = Config({"depot": {"remote": remote.url}}, environ={})
+    fetcher = CatalogFetcher(tmp_path / "fresh", cfg)
+    catalog, locations = fetcher.catalog(version["id"])
+    assert catalog.sources()["remote"] == body
+    assert locations