From 73ca1297cafbe8db01c3bc288ab0b9450e796e82 Mon Sep 17 00:00:00 2001 From: nichinichisou Date: Sun, 18 Oct 2026 20:12:23 +0800 Subject: [PATCH] feat(depot): share catalog versions through remotes --- CHANGELOG.md | 1 + docs/contracts.md | 11 +++- src/nnnotes/cli/assets.py | 5 +- src/nnnotes/cli/catalog.py | 4 +- src/nnnotes/pipeline/catalogdb.py | 63 ++++++++++++++++++-- src/nnnotes/pipeline/depot.py | 7 ++- src/nnnotes/pipeline/remote.py | 6 +- tests/cli/test_assets.py | 28 +++++++++ tests/pipeline/test_catalog_remote.py | 85 +++++++++++++++++++++++++++ 9 files changed, 194 insertions(+), 16 deletions(-) create mode 100644 tests/pipeline/test_catalog_remote.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ac9277d..91c8ea9 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/docs/contracts.md b/docs/contracts.md index d323c51..ed0bf6c 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -295,6 +295,8 @@ ac/ab/.json results inputs//ab/.json input identities by name runs/.json run manifests runs/selections/ the id of the latest run of a selection +catalogs/versions/.json one catalog version and its labels +catalogs/.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 @@ -302,10 +304,15 @@ remote and kept; a content read from the remote is checked against its content i 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 diff --git a/src/nnnotes/cli/assets.py b/src/nnnotes/cli/assets.py index d36ca5a..25f7c4d 100644 --- a/src/nnnotes/cli/assets.py +++ b/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: diff --git a/src/nnnotes/cli/catalog.py b/src/nnnotes/cli/catalog.py index 67273b6..d49fd60 100644 --- a/src/nnnotes/cli/catalog.py +++ b/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: diff --git a/src/nnnotes/pipeline/catalogdb.py b/src/nnnotes/pipeline/catalogdb.py index fd3655d..01b37b5 100644 --- a/src/nnnotes/pipeline/catalogdb.py +++ b/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/.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 (/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, diff --git a/src/nnnotes/pipeline/depot.py b/src/nnnotes/pipeline/depot.py index 0bd5b5b..5d7dfc9 100644 --- a/src/nnnotes/pipeline/depot.py +++ b/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) diff --git a/src/nnnotes/pipeline/remote.py b/src/nnnotes/pipeline/remote.py index ffa50c6..9e89c9a 100644 --- a/src/nnnotes/pipeline/remote.py +++ b/src/nnnotes/pipeline/remote.py @@ -7,12 +7,14 @@ sha256 of the name) /runs/.json run manifests /runs/selections/ the id of the latest run of a selection + /catalogs/versions/.json one catalog version and its labels + /catalogs/.bin catalog bytes, verified when fetched A remote is `s3:///` (S3-compatible object storage: path-style requests, checksums only where the request needs them, so stores other than AWS answer them) or `file://` (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 diff --git a/tests/cli/test_assets.py b/tests/cli/test_assets.py index e3e29f1..1f6175c 100644 --- a/tests/cli/test_assets.py +++ b/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") diff --git a/tests/pipeline/test_catalog_remote.py b/tests/pipeline/test_catalog_remote.py new file mode 100644 index 0000000..30f4736 --- /dev/null +++ b/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