diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0b8bf00..f6893a7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -32,7 +32,7 @@ jobs: - name: Install (with the deck model and the fonts extra) run: | python -m pip install --upgrade pip - python -m pip install -e ".[test,deck,fonts]" + python -m pip install -e ".[test,deck,fonts,remote]" - name: Lint run: | python -m ruff check . @@ -55,7 +55,7 @@ jobs: - name: Install run: | python -m pip install --upgrade pip - python -m pip install -e ".[test,deck,fonts]" + python -m pip install -e ".[test,deck,fonts,remote]" - name: Type check run: python -m pyright diff --git a/docs/assets.md b/docs/assets.md index a5471a4..1eeb842 100644 --- a/docs/assets.md +++ b/docs/assets.md @@ -126,7 +126,9 @@ Removes what no kept run reaches. The kept runs are, for each selection, the `-- the latest run of a selection is always kept). A kept run reaches the results of its tasks, the contents those results name (their artifacts, and the inputs their task keys name when the depot holds them) and its layout manifests with the files they list. Every other run, result (with its measured cost) and content is removed. -`--dry-run` counts what would be removed. A layout made of hard links or clones keeps its files. gc runs only when +`--dry-run` counts what would be removed. A layout made of hard links or clones keeps its files. `--remote` +collects the depot's remote the same way, by the runs on the remote; its results and contents younger than a day stay, +since a run on another machine may be writing them. gc runs only when no other command uses the depot (it exits with an error otherwise), and commands that start meanwhile wait for it. The downloads are bounded separately, by `[depot] downloads` ([commands.md](commands.md#downloads)). diff --git a/docs/cli.md b/docs/cli.md index abe2ac7..469c685 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -356,6 +356,7 @@ remove the runs, results and contents no kept run reaches | --- | --- | | `--keep` `N` | keep the N latest runs of each selection (default: every run; the latest run of a selection is kept) | | `--dry-run` | count what would be removed, remove nothing | +| `--remote` | also collect the depot's remote, by its runs (its results and contents younger than a day stay) | ## nnnotes voices list diff --git a/docs/configuration.md b/docs/configuration.md index da9b42f..532e9b9 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -111,6 +111,11 @@ exist when they are given. | `[paths] fonts.` | `NNNOTES_PATHS_FONTS_` (e.g. `NNNOTES_PATHS_FONTS_ZH_HANT`) | `--font =PATH` (of `site build`) | font file (OpenType or TrueType) the story text of a language is drawn with (`--fonts open`); also read for the other story languages whose game fonts fall back to that language's font | | `[paths] fonts.emoji` | `NNNOTES_PATHS_FONTS_EMOJI` | `--font emoji=PATH` (of `site build`) | colour emoji font (PNG bitmap glyphs: CBDT or sbix) the stories' emoji sprites are drawn from (`--fonts open`); Noto Color Emoji (SIL Open Font License 1.1) is the tested font. Optional: without it the sprites keep their layout with empty glyphs, listed in `ui/fonts.json` `coverage.sprites.missing` | | `[depot] downloads` | `NNNOTES_DEPOT_DOWNLOADS` | — | how much of the depot's downloaded bundles and raw files to keep once no command uses the depot: a size (`500MB`, `20GiB`; units B, KB, MB, GB, TB and KiB, MiB, GiB, TiB), `all`, or `0` (unset) ([commands.md](commands.md#downloads)) | +| `[depot] remote` | `NNNOTES_DEPOT_REMOTE` | — | a remote of the depot shared by the machines that use it: `s3:///` (S3-compatible object storage; `pip install 'nnnotes[remote]'`) or `file:///` ([contracts.md](contracts.md#depot-remote)) | +| `[depot] endpoint` | `NNNOTES_DEPOT_ENDPOINT` | — | endpoint URL of the S3-compatible object storage; unset: AWS S3 | +| `[depot] access_key_id` | `NNNOTES_DEPOT_ACCESS_KEY_ID` | — | access key id of the object storage; unset: boto3 looks for credentials itself (`AWS_ACCESS_KEY_ID`, `~/.aws`) | +| `[depot] secret_access_key` | `NNNOTES_DEPOT_SECRET_ACCESS_KEY` | — | secret access key of the object storage | +| `[depot] signing_region` | `NNNOTES_DEPOT_SIGNING_REGION` | — | region the requests to the object storage are signed for; unset: `us-east-1` | | `[export] link` | `NNNOTES_EXPORT_LINK` | `--link` (of `export assets`) | how `export assets` makes the layout files from the depot: `auto` (unset), `clone`, `hard` or `copy` ([assets.md](assets.md)) | Hex values are case-insensitive; a `0x` prefix and surrounding whitespace are ignored. All values are TOML strings diff --git a/docs/contracts.md b/docs/contracts.md index 7a7bb3a..9a9cb65 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -256,5 +256,29 @@ The depot is a directory: the downloaded game files, and the results of all stag Every file is written to a temporary name and renamed into place, so concurrent writers never see or leave a partial file and an interrupted task leaves no result; a record that changes is one file, replaced whole. `cas/` and -`ac/` depend only on the tasks run: two runs of the same tasks, with any number of workers, leave identical trees. -`downloads/`, `inputs/`, `runs/*.log.jsonl`, `costs/` and `leases/` are local and not deterministic. +`ac/` depend only on the tasks run: two runs of the same tasks, with any number of workers, leave identical trees; +`inputs/`, `runs/`, `costs/` and `leases/` are not deterministic. + +### Depot remote + +`[depot] remote` gives the depot a remote shared by the machines that use it: `s3:///` in +S3-compatible object storage (`pip install 'nnnotes[remote]'`; path-style requests, so stores other than AWS answer +them) or `file:///`. The remote holds the same keys as the depot directory, below its prefix: + +``` +cas/sha256/ab/ contents +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 +``` + +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 +stay in the depot directory. + +Every request names one object; only `depot gc --remote` lists the remote. diff --git a/pyproject.toml b/pyproject.toml index 57a1288..a587f65 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,6 +29,8 @@ dynamic = ["version"] test = ["pytest>=8", "ruff>=0.15", "jsonschema>=4.18", "import-linter>=2.1", "pyright>=1.1.414"] fonts = ["fonttools>=4.50", "freetype-py>=2.5", "scipy>=1.13"] docs = ["mkdocs>=1.6,<2", "mkdocs-material>=9.5", "mkdocstrings[python]>=0.29"] +# a depot remote in S3-compatible object storage ([depot] remote = "s3://...") +remote = ["boto3>=1.35"] # the deck model of music-data: one nnnotes version carries one ournotes-deck release (.github/workflows/deck.yml moves # the pin when ournotes-deck publishes a newer one) deck = ["ournotes-sim==0.0.8"] diff --git a/src/nnnotes/cli/assets.py b/src/nnnotes/cli/assets.py index 6c336d8..0c4eef3 100644 --- a/src/nnnotes/cli/assets.py +++ b/src/nnnotes/cli/assets.py @@ -456,13 +456,13 @@ class Workspace: def __init__(self, args, cfg, log=_log): from ..pipeline.catalogdb import CatalogDB - from ..pipeline.depot import Depot + from ..pipeline.depot import open_depot self.args, self.cfg, self.log = args, cfg, log self.root = Session(cfg).depot_root() self.apk = cfg.path("paths", "apk") self.fetcher = CatalogFetcher(self.root, cfg) - self.depot = Depot(self.root, fetch=self.fetcher) + self.depot = open_depot(cfg, fetch=self.fetcher) self.db = CatalogDB(self.root) try: self.selection = parse_selection(getattr(args, "select", None)) @@ -1399,7 +1399,7 @@ def _task_files(items) -> list[Path]: def cmd_run_stage(args, cfg): - from ..pipeline.depot import Depot + from ..pipeline.depot import open_depot from ..pipeline.orchestrate import run_one problems, tasks = [], [] @@ -1425,7 +1425,7 @@ def cmd_run_stage(args, cfg): print(f"nnnotes: incompatible task: {m}", file=sys.stderr) sys.exit(EXIT_INCOMPATIBLE) root = Session(cfg).depot_root() - depot = Depot(root, fetch=CatalogFetcher(root, cfg) if args.fetch else None) + depot = open_depot(cfg, fetch=CatalogFetcher(root, cfg) if args.fetch else None) use(cfg) out, bad = [], 0 for t in tasks: diff --git a/src/nnnotes/cli/catalog.py b/src/nnnotes/cli/catalog.py index f1f6f14..67273b6 100644 --- a/src/nnnotes/cli/catalog.py +++ b/src/nnnotes/cli/catalog.py @@ -190,7 +190,7 @@ def cmd_fetch(args, cfg): def cmd_diff(args, cfg): from ..pipeline import catalogdb - from ..pipeline.depot import Depot + from ..pipeline.depot import open_depot from .assets import addresses_of db = _db(cfg) @@ -200,7 +200,7 @@ def cmd_diff(args, cfg): vs.append(db.get(ref)) except (KeyError, ValueError) as e: usage(args, f"{ref}: {e.args[0] if e.args else e}") - depot = Depot(Session(cfg).depot_root()) + depot = open_depot(cfg) a, b = vs d = catalogdb.diff(db.index(a), db.index(b), addresses_of(depot, a["id"]), addresses_of(depot, b["id"])) if args.json: diff --git a/src/nnnotes/cli/depot.py b/src/nnnotes/cli/depot.py index d7f5ba2..1e3a431 100644 --- a/src/nnnotes/cli/depot.py +++ b/src/nnnotes/cli/depot.py @@ -23,7 +23,7 @@ def cmd_verify(args, cfg): def cmd_gc(args, cfg): from ..core.downloads import exclusive - from ..pipeline.depot import Depot + from ..pipeline.depot import open_depot if args.keep is not None and args.keep < 1: usage(args, "--keep must be at least 1") @@ -34,17 +34,20 @@ def cmd_gc(args, cfg): f"nnnotes depot gc: another command uses the depot ({root / 'leases'}); " "gc runs when no other command does" ) - r = Depot(root).gc(keep=args.keep, dry_run=args.dry_run) - report( - args, - r, - lambda r: [ - f"{'would remove' if args.dry_run else 'removed'}: {r['runs']['removed']} runs, " + depot = open_depot(cfg) + if args.remote and depot.remote is None: + usage(args, "--remote: the depot has no remote (setting depot.remote)") + r = depot.gc(keep=args.keep, dry_run=args.dry_run, remote=args.remote) + + def lines(r: dict, where: str = "") -> list[str]: + return [ + f"{where}{'would remove' if args.dry_run else 'removed'}: {r['runs']['removed']} runs, " f"{r['results']['removed']} results, {r['contents']['removed']} contents " f"({r['bytesRemoved'] / 1e6:.1f} MB)", - f"kept: {r['runs']['kept']} runs, {r['results']['kept']} results, {r['contents']['kept']} contents", - ], - ) + f"{where}kept: {r['runs']['kept']} runs, {r['results']['kept']} results, {r['contents']['kept']} contents", + ] + + report(args, r, lambda r: lines(r) + (lines(r["remote"], "remote ") if "remote" in r else [])) def register(sub) -> None: @@ -81,3 +84,8 @@ def register(sub) -> None: help="keep the N latest runs of each selection (default: every run; the latest run of a selection is kept)", ) c.add_argument("--dry-run", action="store_true", help="count what would be removed, remove nothing") + c.add_argument( + "--remote", + action="store_true", + help="also collect the depot's remote, by its runs (its results and contents younger than a day stay)", + ) diff --git a/src/nnnotes/pipeline/cristages.py b/src/nnnotes/pipeline/cristages.py index 8d4d794..c80c897 100644 --- a/src/nnnotes/pipeline/cristages.py +++ b/src/nnnotes/pipeline/cristages.py @@ -92,7 +92,7 @@ def boot_input(depot, apk) -> Input | None: if known is None or not depot.has(*known): with ApkSet(apk) as z: data = z.read(BOOT_IN_APK) - known = depot.identify(depot.path(depot.put(data)), "files", name) + known = depot.identify(depot.local(depot.put(data)), "files", name) return Input("boot", known[0], known[1], "data.unity3d", ({"kind": "depot"},)) diff --git a/src/nnnotes/pipeline/depot.py b/src/nnnotes/pipeline/depot.py index 86d0184..0bd5b5b 100644 --- a/src/nnnotes/pipeline/depot.py +++ b/src/nnnotes/pipeline/depot.py @@ -25,6 +25,11 @@ The files of cas/ are read-only (a layout may hard-link them: an edit of such a file fails instead of changing the depot). Only costs/, runs/ and inputs/ hold non-deterministic data; cas/ and ac/ are a function of the tasks run. + +A depot may have a remote (pipeline.remote): object storage or a directory shared by the machines that use it, with +the same keys as the files above (cas/, ac/, inputs//, runs/). The local depot is its cache: a content, +result, input identity or run it lacks is read from the remote and kept, and what it adds goes to the remote as well +(the bytes of a result before the result). downloads/, inputs/paths/, catalogs/, costs/ and run logs stay local. """ from __future__ import annotations @@ -40,8 +45,11 @@ from ..core.downloads import touch from . import contract from .contract import Input +from .remote import Remote, open_remote INPUT_KINDS = ("bundles", "raw", "files") +REMOTE_GRACE = 24 * 3600 # `gc(remote=True)` keeps results and contents of the remote younger than this (seconds) +DONE = ("ran", "hit") # the statuses of a run's tasks whose results are committed class InputMissing(FileNotFoundError): @@ -151,26 +159,62 @@ def _record_path(directory: Path, key: str) -> Path: class Depot: """A depot directory. `fetch(input, locator) -> Path`: how a "catalog" locator is fetched (none: such locators - are skipped).""" + are skipped). `remote`: the depot's remote (None: a local depot).""" - def __init__(self, root, *, fetch=None): + def __init__(self, root, *, fetch=None, remote: Remote | None = None): self.root = Path(root) self.fetch = fetch + self.remote = remote self.written = self.reused = 0 - self._have: set[str] = set() # content ids this depot object wrote or found + self._have: set[str] = set() # content ids this depot object wrote or found, here or on the remote + self._sizes: dict[str, int] = {} # remote key -> size, of the objects known to be on the remote + self._listed: set[str] = set() # task keys whose results the remote holds (the latest runs read from it) self._hashed: dict[str, dict] = {} # path -> its record (inputs/paths) self._names: dict[tuple[str, str], tuple[str, int]] = {} # (kind, name) -> (sha256, size) self._docs: dict[str, dict] = {} # the documents document() decoded last self._lock = threading.Lock() def __getstate__(self): # a depot crosses to worker processes as its settings - return {"root": self.root, "fetch": self.fetch} + return {"root": self.root, "fetch": self.fetch, "remote": self.remote} def __setstate__(self, state): - self.__init__(state["root"], fetch=state["fetch"]) + self.__init__(state["root"], fetch=state["fetch"], remote=state["remote"]) def __repr__(self) -> str: - return f"Depot({str(self.root)!r})" + more = f", remote={self.remote.url!r}" if self.remote is not None else "" + return f"Depot({str(self.root)!r}{more})" + + def _key(self, p: Path) -> str: + """The remote key of a file of the depot: its path below the root.""" + return p.relative_to(self.root).as_posix() + + def _on_remote(self, key: str, size: int | None = None) -> bool: + """Whether the remote holds the object `key` (of `size`).""" + n = self._sizes.get(key) + if n is None: + n = self.remote.size(key) if self.remote is not None else None + if n is None: + return False + self._sizes[key] = n + return size is None or n == size + + def _push(self, sha: str, size: int) -> None: + """Copy a content of this depot to the remote unless the remote holds it.""" + if self.remote is not None: + p = self.path(sha) + key = self._key(p) + if not self._on_remote(key, size): + self.remote.put_file(key, p) + self._sizes[key] = size + + def _fetch_record(self, p: Path) -> bytes | None: + """The bytes of a record file of the depot, read from the remote (and kept) when only the remote holds it.""" + data = _read(p) + if data is None and self.remote is not None: + data = self.remote.get(self._key(p)) + if data is not None: + _write(p, data, lambda q: True) + return data @property def downloads(self) -> Path: @@ -189,30 +233,45 @@ def path(self, sha: str) -> Path: return self.root / "cas" / "sha256" / sha[:2] / sha def has(self, sha: str, size: int | None = None) -> bool: + """Whether the depot or its remote holds the content (of `size`).""" if sha in self._have: return True p = self.path(sha) try: n = p.stat().st_size except OSError: - return False - if size is not None and n != size: - return False + if not self._on_remote(self._key(p), size): + return False + else: + if size is not None and n != size: + return False self._have.add(sha) return True + def local(self, sha: str) -> Path: + """The file of a content in this depot, read from the remote first when only the remote holds it.""" + p = self.path(sha) + if self.remote is not None and not p.is_file(): + data = self.remote.get(self._key(p)) + if data is None: + raise FileNotFoundError(f"content {sha} is not in the depot or its remote") + if contract.sha256(data) != sha: + raise RuntimeError(f"content {sha} of the remote differs from its id") + _write(p, data, lambda q: q.stat().st_size == len(data), read_only=True) + return p + def put(self, data) -> str: """Store bytes; their content id. Write-once: bytes already stored are not written again.""" data = bytes(data) sha = contract.sha256(data) if self.has(sha, len(data)): self.reused += 1 - return sha - if _write(self.path(sha), data, lambda p: p.stat().st_size == len(data), read_only=True): + elif _write(self.path(sha), data, lambda p: p.stat().st_size == len(data), read_only=True): self.written += 1 else: self.reused += 1 self._have.add(sha) + self._push(sha, len(data)) return sha def put_file(self, src) -> tuple[str, int]: @@ -221,6 +280,7 @@ def put_file(self, src) -> tuple[str, int]: sha, size = self.identify(src) if self.has(sha, size): self.reused += 1 + self._push(sha, size) return sha, size dst = self.path(sha) dst.parent.mkdir(parents=True, exist_ok=True) @@ -241,6 +301,7 @@ def put_file(self, src) -> tuple[str, int]: if _replace(tmp, dst, lambda p: p.stat().st_size == size): self.written += 1 self._have.add(sha) + self._push(sha, size) return sha, size def add(self, data, ext: str, media: str | None = None) -> dict: @@ -277,11 +338,12 @@ def add_chunks(self, chunks, ext: str, media: str | None = None) -> dict: else: self.reused += 1 self._have.add(sha) + self._push(sha, size) ext = ext.lower().lstrip(".") return {"sha256": sha, "size": size, "mediaType": media or contract.media_type(ext), "ext": ext} def read(self, sha: str) -> bytes: - return self.path(sha).read_bytes() + return self.local(sha).read_bytes() def document(self, sha: str) -> dict: """The JSON document stored as `sha`, decoded; the last two decoded are kept (a catalog index is read at @@ -304,7 +366,7 @@ def result_path(self, key: str) -> Path: def result(self, key: str) -> dict | None: """The committed result of a task key, or None.""" - data = _read(self.result_path(key)) + data = self._fetch_record(self.result_path(key)) return None if data is None else contract.loads(data) def committed(self, key: str, task: str) -> dict: @@ -315,7 +377,8 @@ def committed(self, key: str, task: str) -> dict: return doc def has_result(self, key: str) -> bool: - return self.result_path(key).is_file() + p = self.result_path(key) + return p.is_file() or key in self._listed or self._on_remote(self._key(p)) def commit(self, doc: dict) -> bool: """Write a result entry, after checking that the bytes of every artifact it names are stored. Write-once: an @@ -326,6 +389,7 @@ def commit(self, doc: dict) -> bool: c = a["content"] if not self.has(c["sha256"], c["size"]): raise RuntimeError(f"result {doc['key']}: bytes of {a['id']} ({c['sha256']}) are not stored") + self._push(c["sha256"], c["size"]) data = contract.encode(doc) dst = self.result_path(doc["key"]) @@ -337,9 +401,16 @@ def same(p: Path) -> bool: raise DepotConflict(f"result {doc['key']} differs from the one in the depot ({dst})") return True - if same(dst): - return False - return _write(dst, data, same) + here = same(dst) + if self.remote is not None: + key = self._key(dst) + old = self.remote.get(key) + if old is None: + self.remote.put(key, data) + elif old != data: + raise DepotConflict(f"result {doc['key']} differs from the one in the depot's remote") + self._sizes[key] = len(data) + return False if here else _write(dst, data, same) # ---------------------------------------------------------------- inputs def identify(self, path, kind: str = "files", name: str | None = None) -> tuple[str, int]: @@ -364,8 +435,11 @@ def identify(self, path, kind: str = "files", name: str | None = None) -> tuple[ self._hashed[key] = rec hit = (rec["sha256"], rec["size"]) if name is not None and self.named(kind, name) != hit: - doc = {"name": name, "sha256": hit[0], "size": hit[1]} - _write(_record_path(self.root / "inputs" / kind, name), contract.encode(doc)) + doc = contract.encode({"name": name, "sha256": hit[0], "size": hit[1]}) + rec_path = _record_path(self.root / "inputs" / kind, name) + _write(rec_path, doc) + if self.remote is not None: + self.remote.put(self._key(rec_path), doc) with self._lock: self._names[(kind, name)] = hit return hit @@ -375,7 +449,7 @@ def named(self, kind: str, name: str) -> tuple[str, int] | None: with self._lock: hit = self._names.get((kind, name)) if hit is None: - data = _read(_record_path(self.root / "inputs" / kind, name)) + data = self._fetch_record(_record_path(self.root / "inputs" / kind, name)) if data is None: return None doc = json.loads(data) @@ -393,7 +467,7 @@ def open_input(self, inp: Input) -> Path: try: if kind == "depot": if self.has(inp.sha256, inp.size): - return self.path(inp.sha256) + return self.local(inp.sha256) tried.append("depot: absent") continue if kind == "catalog": @@ -494,25 +568,34 @@ def write_run(self, doc: dict, log: list[dict] | None = None) -> Path: if log is not None: text = "".join(json.dumps(line, ensure_ascii=False, sort_keys=True) + "\n" for line in log) write_file(p.with_name(f"{doc['id']}.log.jsonl"), text.encode("utf-8")) - write_file(self._selection_path(doc["selection"]), (doc["id"] + "\n").encode("ascii")) + pointer = self._selection_path(doc["selection"]) + write_file(pointer, (doc["id"] + "\n").encode("ascii")) + if self.remote is not None: + self.remote.put(self._key(p), contract.encode(doc)) + self.remote.put(self._key(pointer), (doc["id"] + "\n").encode("ascii")) return p def _selection_path(self, selection) -> Path: return self.root / "runs" / "selections" / contract.digest(list(selection)) def run(self, run_id: str) -> dict | None: - try: - return contract.loads(self.run_path(run_id).read_bytes()) - except OSError: - return None + data = self._fetch_record(self.run_path(run_id)) + return None if data is None else contract.loads(data) def latest_run(self, selection) -> dict | None: - """The latest run manifest written for `selection`, or None.""" - try: - rid = self._selection_path(selection).read_text(encoding="ascii").strip() - except OSError: - return None - return self.run(rid) + """The latest run manifest written for `selection` (by any machine of the remote), or None. The results of + the tasks of a run read from the remote are known to be on the remote (gc keeps them).""" + pointer = self._selection_path(selection) + data = self.remote.get(self._key(pointer)) if self.remote is not None else None + shared = data is not None + if data is None: + data = _read(pointer) + if data is None: + return None + doc = self.run(data.decode("ascii").strip()) + if shared and doc is not None: + self._listed.update(t["key"] for t in doc.get("tasks") or () if t.get("key") and t.get("status") in DONE) + return doc def write_cost(self, key: str, cost: dict) -> None: write_file(self.root / "costs" / f"{key}.json", contract.encode(cost)) @@ -524,53 +607,51 @@ def cost(self, key: str) -> dict | None: return None # ---------------------------------------------------------------- collection - def gc(self, *, keep: int | None = None, dry_run: bool = False) -> dict: + def gc( + self, *, keep: int | None = None, dry_run: bool = False, remote: bool = False, grace: float = REMOTE_GRACE + ) -> dict: """Remove what no kept run reaches. The kept runs: of each selection the `keep` latest by the time their manifest was written (None: every run), and always the run the selection names as its latest. A kept run reaches the results of its tasks, the contents those results name (their artifacts, and the inputs their key parts name when the depot holds them) and the layout manifests it names with the contents they list. - Removed: the other runs (manifest and log), results and their measured costs, and contents. `dry_run`: count - only. -> {"runs", "results", "contents": {"kept", "removed"}, "bytesRemoved"}.""" + Removed: the other runs (manifest and log), results and their measured costs, and contents. `remote`: the + remote's runs, results and contents too, by the runs of the remote; its results and contents younger than + `grace` seconds stay (a run of another machine may be writing them). `dry_run`: count only. + -> {"runs", "results", "contents": {"kept", "removed"}, "bytesRemoved"}, and "remote": the same.""" + out = self._gc_local(keep, dry_run) + if remote: + if self.remote is None: + raise ValueError("the depot has no remote") + out["remote"] = self._gc_remote(self.remote, keep, dry_run, grace) + return out + + def _gc_local(self, keep: int | None, dry_run: bool) -> dict: runs_dir = self.root / "runs" - manifests = sorted(p for p in runs_dir.glob("*.json") if p.is_file()) latest = set() for p in (runs_dir / "selections").glob("*"): try: latest.add(p.read_text(encoding="ascii").strip()) except OSError: continue - by_selection: dict[str, list[tuple[int, Path, dict]]] = {} - for p in manifests: + runs = [] + for p in sorted(p for p in runs_dir.glob("*.json") if p.is_file()): try: - doc = contract.loads(p.read_bytes()) + runs.append((p.stat().st_mtime_ns, p, contract.loads(p.read_bytes()))) except (OSError, ValueError): continue - sel = contract.digest(list(doc.get("selection") or ())) - by_selection.setdefault(sel, []).append((p.stat().st_mtime_ns, p, doc)) - kept, dropped = [], [] - for runs in by_selection.values(): - runs.sort(key=lambda r: r[0], reverse=True) - for i, (_, p, doc) in enumerate(runs): - (kept if keep is None or i < keep or doc.get("id") in latest else dropped).append((p, doc)) - keys: set[str] = set() - shas: set[str] = set() - for _, doc in kept: - for t in doc.get("tasks") or (): - if t.get("key"): - keys.add(t["key"]) - for sha in (doc.get("layouts") or {}).values(): - shas.add(sha) - if self.has(sha): - _collect_shas(contract.loads(self.read(sha)), shas) - for key in keys: - res = self.result(key) - if res is None: - continue - shas.update(a["content"]["sha256"] for a in res.get("artifacts", ())) - shas.update(sha for _, sha in (res.get("keyParts") or {}).get("inputs", ())) - removed_bytes = 0 + kept, dropped = _kept_runs(runs, latest, keep) + + def result(key: str) -> dict | None: + data = _read(self.result_path(key)) + return None if data is None else contract.loads(data) + + def document(sha: str) -> dict | None: + p = self.path(sha) + return contract.loads(p.read_bytes()) if p.is_file() else None + + keys, shas = _reach([doc for _, _, doc in kept], result, document) out: dict = {"runs": {"kept": len(kept), "removed": len(dropped)}} - for p, _ in dropped: + for _, p, _ in dropped: if not dry_run: p.unlink(missing_ok=True) p.with_name(p.name[: -len(".json")] + ".log.jsonl").unlink(missing_ok=True) @@ -585,7 +666,7 @@ def gc(self, *, keep: int | None = None, dry_run: bool = False) -> dict: p.unlink(missing_ok=True) (self.root / "costs" / f"{key}.json").unlink(missing_ok=True) out["results"] = {"kept": n_kept, "removed": n_removed} - n_kept = n_removed = 0 + n_kept = n_removed = removed_bytes = 0 for p in sorted((self.root / "cas" / "sha256").glob("*/*")): if p.name in shas: n_kept += 1 @@ -599,6 +680,108 @@ def gc(self, *, keep: int | None = None, dry_run: bool = False) -> dict: out["bytesRemoved"] = removed_bytes return out + def _gc_remote(self, remote: Remote, keep: int | None, dry_run: bool, grace: float) -> dict: + now = time.time() + latest, runs = set(), [] + for key, _, mtime in remote.list("runs/"): + data = remote.get(key) + if data is None: + continue + if key.startswith("runs/selections/"): + latest.add(data.decode("ascii").strip()) + elif key.endswith(".json"): + runs.append((mtime, key, contract.loads(data))) + kept, dropped = _kept_runs(runs, latest, keep) + keys, shas = _reach( + [doc for _, _, doc in kept], + self.result, + lambda sha: contract.loads(self.read(sha)) if self.has(sha) else None, + ) + out: dict = {"runs": {"kept": len(kept), "removed": len(dropped)}} + gone = [key for _, key, _ in dropped] + counts = {} + removed_bytes = 0 + for part, prefix in (("results", "ac/"), ("contents", "cas/sha256/")): + n_kept = n_removed = 0 + for key, size, mtime in remote.list(prefix): + name = key.rsplit("/", 1)[-1] + reached = name[: -len(".json")] in keys if part == "results" else name in shas + if reached or now - mtime < grace: + n_kept += 1 + continue + n_removed += 1 + gone.append(key) + if part == "contents": + removed_bytes += size + counts[part] = {"kept": n_kept, "removed": n_removed} + if not dry_run: + remote.delete(gone) + for key in gone: + self._sizes.pop(key, None) + self._have.clear() + self._listed.clear() + out.update(counts) + out["bytesRemoved"] = removed_bytes + return out + + +def open_depot(cfg, *, fetch=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.""" + from ..source.config import ConfigError + from ..source.session import Session + + root = Session(cfg).depot_root() + url = cfg.get("depot", "remote") + if not url: + return Depot(root, fetch=fetch) + try: + remote = open_remote( + url, + endpoint=cfg.get("depot", "endpoint"), + key_id=cfg.get("depot", "access_key_id"), + secret=cfg.get("depot", "secret_access_key"), + region=cfg.get("depot", "signing_region"), + ) + except ValueError as e: + raise ConfigError(f"setting depot.remote: {e}") from None + return Depot(root, fetch=fetch, remote=remote) + + +def _kept_runs(runs: list, latest: set[str], keep: int | None) -> tuple[list, list]: + """(kept, dropped) of runs (time, handle, manifest): of each selection the `keep` latest by time (None: every + run), and the runs `latest` names.""" + by_selection: dict[str, list] = {} + for r in runs: + by_selection.setdefault(contract.digest(list(r[2].get("selection") or ())), []).append(r) + kept, dropped = [], [] + for rs in by_selection.values(): + rs.sort(key=lambda r: r[0], reverse=True) + for i, r in enumerate(rs): + (kept if keep is None or i < keep or r[2].get("id") in latest else dropped).append(r) + return kept, dropped + + +def _reach(runs: list[dict], result, document) -> tuple[set[str], set[str]]: + """(task keys, content ids) that run manifests reach: their tasks' results with the contents those name, and + their layout manifests with the contents they list. `result(key)`, `document(sha)`: None when absent.""" + keys: set[str] = set() + shas: set[str] = set() + for doc in runs: + keys.update(t["key"] for t in doc.get("tasks") or () if t.get("key")) + for sha in (doc.get("layouts") or {}).values(): + shas.add(sha) + manifest = document(sha) + if manifest is not None: + _collect_shas(manifest, shas) + for key in keys: + res = result(key) + if res is None: + continue + shas.update(a["content"]["sha256"] for a in res.get("artifacts", ())) + shas.update(sha for _, sha in (res.get("keyParts") or {}).get("inputs", ())) + return keys, shas + def _collect_shas(doc, out: set[str]) -> None: """Add every "sha256" value of a document (a layout manifest names its files' contents so).""" diff --git a/src/nnnotes/pipeline/layout.py b/src/nnnotes/pipeline/layout.py index 1ef6591..cc9a3c2 100644 --- a/src/nnnotes/pipeline/layout.py +++ b/src/nnnotes/pipeline/layout.py @@ -367,8 +367,8 @@ def paths(entries_: list[dict]) -> dict[str, str]: # ---------------------------------------------------------------- writing def from_depot(depot): - """A byte source reading the depot's cas/.""" - return depot.path + """A byte source reading the depot's cas/ (a content only its remote holds is read from the remote first).""" + return depot.local def from_layout(directory): diff --git a/src/nnnotes/pipeline/remote.py b/src/nnnotes/pipeline/remote.py new file mode 100644 index 0000000..ffa50c6 --- /dev/null +++ b/src/nnnotes/pipeline/remote.py @@ -0,0 +1,233 @@ +"""The remote of a depot: a copy of its contents, results, input identities and runs in object storage, shared by the +machines that use it. + + /cas/sha256/ab/ contents + /ac/ab/.json results + /inputs//.json input identities by name ({"name", "sha256", "size"}; digest: the + sha256 of the name) + /runs/.json run manifests + /runs/selections/ the id of the latest run of a selection + +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). +""" + +from __future__ import annotations + +import os +import threading +import urllib.parse +from collections.abc import Iterator +from pathlib import Path + +from ..core.cache import temp_path + +__all__ = ["Remote", "open_remote"] + + +class Remote: + """The operations of a remote, by object key (a relative path under the remote's root).""" + + url: str + + def get(self, key: str) -> bytes | None: + """The object's bytes, None when it does not exist.""" + raise NotImplementedError + + def size(self, key: str) -> int | None: + """The object's size, None when it does not exist.""" + raise NotImplementedError + + def put(self, key: str, data: bytes) -> None: + """Write (or replace) the object.""" + raise NotImplementedError + + def put_file(self, key: str, path: Path) -> None: + """Write (or replace) the object with a file's bytes.""" + self.put(key, Path(path).read_bytes()) + + def delete(self, keys: list[str]) -> None: + """Remove objects (missing ones are no error).""" + raise NotImplementedError + + def list(self, prefix: str) -> Iterator[tuple[str, int, float]]: + """(key, size, time it was written: seconds since the epoch) of every object whose key starts with + `prefix`.""" + raise NotImplementedError + + def __repr__(self) -> str: + return f"{type(self).__name__}({self.url!r})" + + +class FileRemote(Remote): + """A remote in a directory (`file://`).""" + + def __init__(self, url: str, root: Path): + self.url, self.root = url, Path(root) + + def __getstate__(self): + return {"url": self.url, "root": self.root} + + def __setstate__(self, state): + self.__init__(state["url"], state["root"]) + + def _path(self, key: str) -> Path: + parts = key.split("/") + if not key or any(p in ("", ".", "..") for p in parts): + raise ValueError(f"not an object key: {key!r}") + return self.root.joinpath(*parts) + + def get(self, key: str) -> bytes | None: + try: + return self._path(key).read_bytes() + except FileNotFoundError: + return None + + def size(self, key: str) -> int | None: + try: + return self._path(key).stat().st_size + except FileNotFoundError: + return None + + def put(self, key: str, data: bytes) -> None: + dst = self._path(key) + dst.parent.mkdir(parents=True, exist_ok=True) + tmp = temp_path(dst) + try: + with open(tmp, "xb") as f: + f.write(data) + os.replace(tmp, dst) + except BaseException: + tmp.unlink(missing_ok=True) + raise + + def delete(self, keys: list[str]) -> None: + for k in keys: + self._path(k).unlink(missing_ok=True) + + def list(self, prefix: str) -> Iterator[tuple[str, int, float]]: + if not self.root.is_dir(): + return + for p in sorted(self.root.rglob("*")): + if p.is_file() and not p.name.endswith(".part"): + key = p.relative_to(self.root).as_posix() + if key.startswith(prefix): + st = p.stat() + yield key, st.st_size, st.st_mtime + + +class S3Remote(Remote): + """A remote in S3-compatible object storage (`s3:///`), through boto3 (the `remote` extra).""" + + def __init__(self, url: str, endpoint: str | None, key_id: str | None, secret: str | None, region: str): + parts = urllib.parse.urlsplit(url) + if not parts.netloc: + raise ValueError("an s3:// remote names its bucket: s3:///") + self.url, self.bucket = url, parts.netloc + prefix = parts.path.strip("/") + self.prefix = f"{prefix}/" if prefix else "" + self.endpoint, self.key_id, self.secret, self.region = endpoint, key_id, secret, region + self._local = threading.local() + + def __getstate__(self): + return {k: getattr(self, k) for k in ("url", "endpoint", "key_id", "secret", "region")} + + def __setstate__(self, state): + self.__init__(state["url"], state["endpoint"], state["key_id"], state["secret"], state["region"]) + + def __repr__(self) -> str: # never the credentials + return f"S3Remote({self.url!r})" + + @property + def _s3(self): + """One client per thread (boto3 clients are not shared between threads here).""" + client = getattr(self._local, "client", None) + if client is None: + try: + import boto3 + from botocore.config import Config + except ImportError: + raise RuntimeError("an s3:// depot remote needs boto3: pip install 'nnnotes[remote]'") from None + config = Config( + s3={"addressing_style": "path"}, + retries={"max_attempts": 8, "mode": "adaptive"}, + request_checksum_calculation="when_required", + response_checksum_validation="when_required", + ) + client = self._local.client = boto3.client( + "s3", + endpoint_url=self.endpoint or None, + aws_access_key_id=self.key_id or None, + aws_secret_access_key=self.secret or None, + region_name=self.region, + config=config, + ) + return client + + @staticmethod + def _missing(error) -> bool: + return error.response.get("Error", {}).get("Code") in ("404", "NoSuchKey", "NotFound") + + def get(self, key: str) -> bytes | None: + from botocore.exceptions import ClientError + + try: + return self._s3.get_object(Bucket=self.bucket, Key=self.prefix + key)["Body"].read() + except ClientError as e: + if self._missing(e): + return None + raise + + def size(self, key: str) -> int | None: + from botocore.exceptions import ClientError + + try: + return int(self._s3.head_object(Bucket=self.bucket, Key=self.prefix + key)["ContentLength"]) + except ClientError as e: + if self._missing(e): + return None + raise + + def put(self, key: str, data: bytes) -> None: + self._s3.put_object(Bucket=self.bucket, Key=self.prefix + key, Body=data) + + def put_file(self, key: str, path: Path) -> None: + self._s3.upload_file(str(path), self.bucket, self.prefix + key) + + def delete(self, keys: list[str]) -> None: + for i in range(0, len(keys), 1000): + batch = [{"Key": self.prefix + k} for k in keys[i : i + 1000]] + if batch: + self._s3.delete_objects(Bucket=self.bucket, Delete={"Objects": batch, "Quiet": True}) + + def list(self, prefix: str) -> Iterator[tuple[str, int, float]]: + pages = self._s3.get_paginator("list_objects_v2").paginate(Bucket=self.bucket, Prefix=self.prefix + prefix) + for page in pages: + for o in page.get("Contents", ()): + yield o["Key"][len(self.prefix) :], int(o["Size"]), o["LastModified"].timestamp() + + +def open_remote( + url: str, + *, + endpoint: str | None = None, + key_id: str | None = None, + secret: str | None = None, + region: str | None = None, +) -> Remote: + """The remote a URL names: `s3:///` (with the endpoint, credentials and signing region of the + object storage; region default us-east-1) or `file://`.""" + parts = urllib.parse.urlsplit(url) + if parts.scheme == "s3": + return S3Remote(url, endpoint, key_id, secret, region or "us-east-1") + if parts.scheme == "file": + path = urllib.parse.unquote(parts.path) + if parts.netloc or not path: # file://host/path is not a local directory + raise ValueError("a file:// remote names a local directory: file:///") + if os.name == "nt" and len(path) > 2 and path[0] == "/" and path[2] == ":": + path = path[1:] # file:///C:/x + return FileRemote(url, Path(path)) + raise ValueError("a depot remote is s3:/// or file:///") diff --git a/src/nnnotes/source/configfile.py b/src/nnnotes/source/configfile.py index 14321b5..4f4774b 100644 --- a/src/nnnotes/source/configfile.py +++ b/src/nnnotes/source/configfile.py @@ -23,7 +23,14 @@ from . import languages from .config import Config, ConfigError, check_catalog_version, env_name, template -SECRETS = {("bundle", "key"), ("bundle", "nonce_seed"), ("master", "key"), ("master", "iv")} +SECRETS = { + ("bundle", "key"), + ("bundle", "nonce_seed"), + ("master", "key"), + ("master", "iv"), + ("depot", "access_key_id"), + ("depot", "secret_access_key"), +} HEX_BYTES = {("bundle", "key"): 16, ("bundle", "nonce_seed"): None, ("master", "key"): 32, ("master", "iv"): 32} REGION = "servers." # the section of the per-region settings TEMPLATE_REGION = "tw" # the example region table of the template @@ -155,6 +162,14 @@ def validate(setting: Setting, value: Value) -> None: raise ValueError(f"must be one of {', '.join(LINKS)}") elif k == ("depot", "downloads"): parse_budget(value) + elif k == ("depot", "remote"): + u = urlsplit(value) + if not ((u.scheme == "s3" and u.netloc) or (u.scheme == "file" and not u.netloc and u.path)): + raise ValueError("must be s3:/// or file:///") + elif k == ("depot", "endpoint"): + u = urlsplit(value) + if u.scheme not in ("https", "http") or not u.netloc: + raise ValueError("must be an http(s):// URL") # ---------------------------------------------------------------- TOML text diff --git a/src/nnnotes/source/nnnotes.example.toml b/src/nnnotes/source/nnnotes.example.toml index 437beb1..350e687 100644 --- a/src/nnnotes/source/nnnotes.example.toml +++ b/src/nnnotes/source/nnnotes.example.toml @@ -81,6 +81,18 @@ player = "" # how much of /downloads (decrypted bundles, raw CDN files) to keep once no command uses the depot: a size such # as 20GiB or 500MB, or all; empty: 0 (the last command to end removes them; they are downloaded again when needed) downloads = "" +# a remote of the depot, shared by the machines that use it: s3:/// (S3-compatible object storage, +# through boto3: pip install 'nnnotes[remote]') or file:///; the depot reads the contents, results, input +# identities and runs it lacks from the remote and writes what it adds to it +remote = "" +# the endpoint URL of the S3-compatible object storage of the remote; empty: AWS S3 +endpoint = "" +# the access key id of the object storage; empty: boto3 looks for credentials itself (AWS_ACCESS_KEY_ID, ~/.aws) +access_key_id = "" +# the secret access key of the object storage +secret_access_key = "" +# the region requests to the object storage are signed for; empty: us-east-1 +signing_region = "" [export] # how `nnnotes export assets` makes the layout files from the depot: auto (a clone where the file system makes them, else diff --git a/tests/cli/test_assets.py b/tests/cli/test_assets.py index f049175..2e10549 100644 --- a/tests/cli/test_assets.py +++ b/tests/cli/test_assets.py @@ -388,6 +388,21 @@ def test_a_depot_that_keeps_no_downloads_hits_without_them(tmp_path, capsys, fak assert again["layouts"]["original"]["write"] == 0 +def test_a_second_depot_with_the_same_remote_hits_without_downloads(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 = export(capsys, tmp_path, d, "b", "out-b") # a depot with no downloads and no CDN + assert out, err + again = json.loads(out) + assert code == 1 and again["tasks"]["ran"] == 0 and again["run"] == first["run"], err + skip = [f"{assets.REPORTS}/run.json"] # the statuses: ran, then hit + assert tree(tmp_path / "out-b", skip) == tree(tmp_path / "out-a", skip) + assert not [p for p in (tmp_path / "b").glob("downloads/**/*") if p.is_file()] # nothing downloaded + + 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_depot_gc.py b/tests/pipeline/test_depot_gc.py index 75df301..7170dd6 100644 --- a/tests/pipeline/test_depot_gc.py +++ b/tests/pipeline/test_depot_gc.py @@ -94,3 +94,14 @@ def test_the_gc_command_runs_only_when_no_other_command_does(tmp_path, capsys): cli.main(argv) assert json.loads(capsys.readouterr().out)["contents"]["removed"] == 1 and not d.path(orphan).exists() assert not list((tmp_path / "leases").iterdir()) + + +def test_the_gc_command_collects_the_remote_with_remote(tmp_path, capsys, monkeypatch): + argv = ["--depot", str(tmp_path / "d"), "depot", "gc", "--remote", "--json"] + with pytest.raises(SystemExit): + cli.main(argv) + assert "the depot has no remote" in capsys.readouterr().err + monkeypatch.setenv("NNNOTES_DEPOT_REMOTE", (tmp_path / "remote").as_uri()) + cli.main(argv) + r = json.loads(capsys.readouterr().out) + assert r["remote"]["runs"] == {"kept": 0, "removed": 0} and r["contents"] == {"kept": 0, "removed": 0} diff --git a/tests/pipeline/test_depot_remote.py b/tests/pipeline/test_depot_remote.py new file mode 100644 index 0000000..fe85851 --- /dev/null +++ b/tests/pipeline/test_depot_remote.py @@ -0,0 +1,173 @@ +"""A depot with a remote: what the local depot writes goes to the remote too, and a depot that lacks something reads it +from the remote and keeps it.""" + +from datetime import UTC, datetime + +import pytest + +from nnnotes.pipeline import contract +from nnnotes.pipeline.contract import Input, Task +from nnnotes.pipeline.depot import Depot, DepotConflict +from nnnotes.pipeline.remote import FileRemote, open_remote + + +def commit(d: Depot, task: Task, payload: bytes) -> dict: + aid = contract.artifact_id(task.id, "data.bin") + art = contract.artifact(aid, d.add(payload, "bin"), contract.provenance(task), {"kind": "test"}) + doc = contract.result(task, [art], []) + d.commit(doc) + return doc + + +def run_doc(selection, tasks) -> dict: + return contract.run_doc( + context={"region": "r"}, + selection=selection, + params={}, + stages={"t.x": 1}, + tasks=[{"id": t.id, "key": t.key, "status": "ran"} for t in tasks], + layouts={}, + summary={}, + ) + + +def test_open_remote_names_a_directory_or_a_bucket(tmp_path): + r = open_remote(tmp_path.as_uri()) + assert isinstance(r, FileRemote) and r.root == tmp_path + s3 = open_remote("s3://bucket/some/prefix", endpoint="https://storage.example", key_id="KEYID", secret="SECRET") + assert (s3.bucket, s3.prefix) == ("bucket", "some/prefix/") + assert "KEYID" not in repr(s3) and "SECRET" not in repr(s3) + for bad in ("http://example/x", "s3:///x", "file://host/share"): + with pytest.raises(ValueError): + open_remote(bad) + + +def test_a_second_depot_finds_the_results_and_contents_of_the_first(tmp_path): + remote = open_remote((tmp_path / "remote").as_uri()) + first, second = Depot(tmp_path / "a", remote=remote), Depot(tmp_path / "b", remote=remote) + stored = first.put(b"a stored input") + task = Task("t.x", 1, "a", inputs=(Input("doc", stored, 14, None, ({"kind": "depot"},)),)) + doc = commit(first, task, b"payload") + run = run_doc(["group:a"], [task]) + first.write_run(run) + first.identify(first.path(stored), "bundles", "x_0123.bundle") + + art = doc["artifacts"][0]["content"]["sha256"] + assert not second.path(art).exists() and second.has(art, 7) and second.has_result(task.key) + assert second.result(task.key) == doc and second.read(art) == b"payload" + assert second.path(art).is_file() # kept locally + assert second.open_input(task.inputs[0]).read_bytes() == b"a stored input" + assert second.latest_run(["group:a"]) == run and second.run(run["id"]) == run + assert second.named("bundles", "x_0123.bundle") == (stored, 14) + assert second.named("bundles", "other.bundle") is None + + +class Counting(FileRemote): + """A directory remote that counts the objects it is asked the size of.""" + + def __init__(self, url, root): + super().__init__(url, root) + self.asked = [] + + def size(self, key): + self.asked.append(key) + return super().size(key) + + +def test_the_latest_run_of_the_remote_tells_which_results_it_holds(tmp_path): + remote = open_remote((tmp_path / "remote").as_uri()) + first = Depot(tmp_path / "a", remote=remote) + tasks = [Task("t.x", 1, n) for n in "abc"] + for t in tasks: + commit(first, t, t.subject.encode()) + first.write_run(run_doc(["group:a"], tasks)) + counting = Counting(remote.url, remote.root) + second = Depot(tmp_path / "b", remote=counting) + assert second.latest_run(["group:a"])["id"] == run_doc(["group:a"], tasks)["id"] + assert all(second.has_result(t.key) for t in tasks) and counting.asked == [] + assert not second.has_result(Task("t.x", 1, "z").key) and len(counting.asked) == 1 + + +def test_a_depot_crosses_processes_with_its_remote(tmp_path): + import pickle + + remote = open_remote((tmp_path / "remote").as_uri()) + d = pickle.loads(pickle.dumps(Depot(tmp_path / "a", remote=remote))) + assert isinstance(d.remote, FileRemote) and d.remote.root == remote.root + assert "remote=" in repr(d) + + +def test_contents_stored_before_the_remote_go_to_it_with_their_result(tmp_path): + task = Task("t.x", 1, "a") + local = Depot(tmp_path / "a") + aid = contract.artifact_id(task.id, "data.bin") + art = contract.artifact(aid, local.add(b"payload", "bin"), contract.provenance(task), {"kind": "test"}) + remote = open_remote((tmp_path / "remote").as_uri()) + Depot(tmp_path / "a", remote=remote).commit(contract.result(task, [art], [])) + assert Depot(tmp_path / "b", remote=remote).read(art["content"]["sha256"]) == b"payload" + + +def test_a_content_of_the_remote_is_checked_against_its_id(tmp_path): + remote = open_remote((tmp_path / "remote").as_uri()) + sha = Depot(tmp_path / "a", remote=remote).put(b"right") + remote.put(f"cas/sha256/{sha[:2]}/{sha}", b"wrong") + with pytest.raises(RuntimeError, match="differs from its id"): + Depot(tmp_path / "b", remote=remote).read(sha) + + +def test_s3_requests_name_the_bucket_and_the_prefixed_keys(): + pytest.importorskip("boto3") + from botocore.stub import Stubber + + r = open_remote("s3://bucket/depot", endpoint="https://storage.example", key_id="k", secret="s") + written = datetime(2026, 1, 2, tzinfo=UTC) + with Stubber(r._s3) as stub: + stub.add_response("head_object", {"ContentLength": 3}, {"Bucket": "bucket", "Key": "depot/ac/ab/x.json"}) + stub.add_client_error( + "head_object", + service_error_code="404", + http_status_code=404, + expected_params={"Bucket": "bucket", "Key": "depot/ac/ab/y.json"}, + ) + stub.add_response("put_object", {}, {"Bucket": "bucket", "Key": "depot/runs/r.json", "Body": b"{}"}) + stub.add_response( + "list_objects_v2", + { + "Contents": [{"Key": "depot/cas/sha256/ab/abc", "Size": 5, "LastModified": written}], + "IsTruncated": False, + }, + {"Bucket": "bucket", "Prefix": "depot/cas/"}, + ) + stub.add_response( + "delete_objects", + {}, + {"Bucket": "bucket", "Delete": {"Objects": [{"Key": "depot/cas/sha256/ab/abc"}], "Quiet": True}}, + ) + assert r.size("ac/ab/x.json") == 3 and r.size("ac/ab/y.json") is None + r.put("runs/r.json", b"{}") + assert list(r.list("cas/")) == [("cas/sha256/ab/abc", 5, written.timestamp())] + r.delete(["cas/sha256/ab/abc"]) + stub.assert_no_pending_responses() + + +def test_a_result_differing_from_the_remote_one_is_a_conflict(tmp_path): + remote = open_remote((tmp_path / "remote").as_uri()) + task = Task("t.x", 1, "a") + commit(Depot(tmp_path / "a", remote=remote), task, b"one") + with pytest.raises(DepotConflict): + commit(Depot(tmp_path / "b", remote=remote), task, b"two") + + +def test_gc_of_the_remote_keeps_what_its_kept_runs_reach(tmp_path): + remote = open_remote((tmp_path / "remote").as_uri()) + d = Depot(tmp_path / "a", remote=remote) + kept, gone = Task("t.x", 1, "a"), Task("t.x", 1, "b") + commit(d, kept, b"kept") + commit(d, gone, b"gone") + d.write_run(run_doc(["group:a"], [kept])) + assert Depot(tmp_path / "b", remote=remote).gc(remote=True)["remote"]["results"] == {"kept": 2, "removed": 0} + r = Depot(tmp_path / "b", remote=remote).gc(remote=True, grace=0) + assert r["remote"]["results"] == {"kept": 1, "removed": 1} and r["remote"]["contents"] == {"kept": 1, "removed": 1} + assert ( + remote.size(f"ac/{kept.key[:2]}/{kept.key}.json") and remote.size(f"ac/{gone.key[:2]}/{gone.key}.json") is None + ) diff --git a/tests/source/test_configfile.py b/tests/source/test_configfile.py index acb20a2..76fa57e 100644 --- a/tests/source/test_configfile.py +++ b/tests/source/test_configfile.py @@ -36,6 +36,29 @@ def leaves(table, section=""): yield f"{section}.{k}" if section else k, v +@pytest.mark.parametrize( + "name, value, ok", + [ + ("depot.remote", "s3://bucket/prefix", True), + ("depot.remote", "file:///srv/depot", True), + ("depot.remote", "s3:///prefix", False), + ("depot.remote", "https://bucket/prefix", False), + ("depot.remote", "file://host/share", False), + ("depot.endpoint", "https://storage.example", True), + ("depot.endpoint", "storage.example", False), + ("depot.downloads", "20GiB", True), + ("depot.downloads", "lots", False), + ], +) +def test_depot_settings_are_checked(name, value, ok): + setting, _ = configfile.resolve(name) + if ok: + configfile.validate(setting, value) + else: + with pytest.raises(ValueError): + configfile.validate(setting, value) + + def test_settings_are_the_documented_ones(): doc = (ROOT / "docs" / "configuration.md").read_text(encoding="utf-8") documented = set() @@ -54,7 +77,14 @@ def test_template_values_are_empty(): data = tomllib.loads(template().decode("utf-8")) assert all(v in ("", []) for _, v in leaves(data)) secrets = {f"{s.section}.{s.key}" for s in configfile.SETTINGS if s.secret} - assert secrets == {"bundle.key", "bundle.nonce_seed", "master.key", "master.iv"} + assert secrets == { + "bundle.key", + "bundle.nonce_seed", + "master.key", + "master.iv", + "depot.access_key_id", + "depot.secret_access_key", + } def test_resolve(): @@ -170,7 +200,7 @@ def ask_secret(p): got = {f"{s}.{k}": v for s, k, v in values} assert regions == ["tw", "kr"] assert got == {"bundle.key": KEY, "master.key": MKEY, "catalog.region": "tw", "catalog.language": "zh-Hant"} - assert len(secret_prompts) == 5 and all("(hidden)" in p for p in secret_prompts) + assert len(secret_prompts) == 7 and all("(hidden)" in p for p in secret_prompts) assert any(p.startswith("[servers.kr] cdn") for p in prompts) text = configfile.render(values, regions) assert set(tomllib.loads(text)["servers"]) == {"tw", "kr"}