diff --git a/oci/CATALOG-REFRESH.md b/oci/CATALOG-REFRESH.md index f87a627b..eaa693c5 100644 --- a/oci/CATALOG-REFRESH.md +++ b/oci/CATALOG-REFRESH.md @@ -9,7 +9,12 @@ a connection failure. Selecting it during a refresh does not start a second work Manual refresh bypasses the conditional check and verifies the complete manifest, allowing it to repair damaged cached files even when the upstream revision has not changed. A repaired snapshot gets a separate immutable directory so existing -readers retain their original snapshot until they return to the menu. +readers retain their original snapshot until they return to the menu. Each +Catalog reader holds an OS file-lock lease. Every worker run, including unchanged +checks and failed downloads, prunes snapshots that are neither a channel's active +snapshot nor leased by an open reader. Reader exit or a crash releases its lease +automatically. Snapshot selection, publication and pruning share a short local +lock; opening a menu never waits for the worker's network lock. The worker checks the latest commit affecting `oci/catalog` in the official MacRimi/ProxMenux repository. GitHub's ETag avoids repeated transfers. When the @@ -46,6 +51,16 @@ acknowledged only after the menu displays it. Exiting during a download therefor preserves its news for the next visit. The new-app count remains visible throughout the current visit and becomes zero on the next visit after acknowledgement. +Downloaded recipes are also checked against the top-level `installer_profile` +keys used by the engine's bundled apps, curated recipes and overlays. A new app +using an unknown key is silently excluded; an existing app retains its previous +compatible, fully composed recipe and its matching settings. Hidden entries do +not contribute to the available/new counters, and retaining an old recipe does +not count it as an update. When a ProxMenux update adds support for that key, the +next background check re-evaluates the catalog even at the same remote revision. +This is a conservative key-presence gate, not validation of new values or nested +options of an already known key. Changes in meaning still require `engine_api`. + Catalog maintainers must increase `catalog/compatibility.json`'s `engine_api` when recipes require incompatible runtime capabilities. The consumer currently supports API 1. Changes to the template schema also require a ProxMenux engine diff --git a/oci/src/proxmenux_oci/catalog.py b/oci/src/proxmenux_oci/catalog.py index e376b07f..aa5d81f2 100644 --- a/oci/src/proxmenux_oci/catalog.py +++ b/oci/src/proxmenux_oci/catalog.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import weakref from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone from pathlib import Path @@ -59,8 +60,10 @@ class Catalog: def __init__(self, root: Path, source: GitHubSource | None = None, *, catalog_dir: Path | None = None) -> None: self.root = root - from .catalog_refresh import current_catalog - self.catalog_dir = catalog_dir if catalog_dir is not None else current_catalog(root) + from .catalog_refresh import pin_catalog + self.catalog_dir, lease = pin_catalog(root, catalog_dir) + if lease is not None: + self._snapshot_lease = weakref.finalize(self, lease.close) self._include_local = catalog_dir is None and self.catalog_dir != root / "catalog" self.local_app_ids: set[str] = set() self.apps_dir = self.catalog_dir / "apps" @@ -825,6 +828,9 @@ class Catalog: if not path.exists(): raise ConversionError(f"No template is available for '{app_id}'") template = json.loads(path.read_text(encoding="utf-8")) + if item.get("_retained_recipe"): + self.validate(template, required=False) + return template self._apply_overlay(item["id"], template) self._apply_category(item["id"], template["catalog_ui"]) self._apply_volume_policy(item["id"], template["container_contract"], diff --git a/oci/src/proxmenux_oci/catalog_refresh.py b/oci/src/proxmenux_oci/catalog_refresh.py index d78a95fc..6bce513d 100644 --- a/oci/src/proxmenux_oci/catalog_refresh.py +++ b/oci/src/proxmenux_oci/catalog_refresh.py @@ -7,6 +7,7 @@ never downloaded. An installed engine always retains its bundled fallback. from __future__ import annotations import argparse +from contextlib import contextmanager from concurrent.futures import ThreadPoolExecutor import fcntl import hashlib @@ -95,6 +96,113 @@ def schema_hash(root): return hashlib.sha256((Path(root) / "schemas/oci-template.schema.json").read_bytes()).hexdigest() +def bundle_hash(root): + # Cheap menu-time engine-update detection; no scan of the bundled recipes. + return hashlib.sha256((Path(root) / "catalog/index.json").read_bytes()).hexdigest() + + +@contextmanager +def snapshot_lock(root): + cache = cache_root(root) + cache.mkdir(parents=True, exist_ok=True) + with (cache / "snapshots.lock").open("a") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + yield + + +def pin_catalog(root, directory=None): + """Select and lease atomically against pruning; no network/worker wait.""" + explicit = directory + directory = Path(directory) if directory is not None else current_catalog(root) + if (directory.parent != cache_root(root) / "snapshots" + or not re.fullmatch(r"[0-9a-f]{40}(?:-[0-9a-f]{12})?", directory.name)): + return directory, None + with snapshot_lock(root): + directory = Path(explicit) if explicit is not None else current_catalog(root) + snapshots = cache_root(root) / "snapshots" + if directory.parent != snapshots or not directory.is_dir(): + return directory, None + leases = cache_root(root) / "leases" + leases.mkdir(exist_ok=True) + lease = (leases / directory.name).open("a") + fcntl.flock(lease, fcntl.LOCK_SH) + return directory, lease + + +def prune_snapshots(root): + """Retain channel pointers and live readers; locks survive PID reuse/crashes.""" + cache = cache_root(root) + with snapshot_lock(root): + keep = {read_json(p).get("snapshot", read_json(p).get("revision")) + for p in cache.glob("*-current.json")} + leases = cache / "leases" + leases.mkdir(exist_ok=True) + for directory in (cache / "snapshots").glob("*"): + if (not SHA.fullmatch(directory.name.split("-")[0]) + or directory.name in keep or directory.is_symlink() or not directory.is_dir()): + continue + lease_path = leases / directory.name + with lease_path.open("a") as lease: + try: + fcntl.flock(lease, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + continue + shutil.rmtree(directory) + lease_path.unlink(missing_ok=True) + + +def supported_profile_keys(root): + keys = set() + for folder in ("apps", "curated", "overlays"): + for path in (Path(root) / "catalog" / folder).glob("*.json"): + value = json.loads(path.read_text()) + keys.update(value.get("proxmox", {}).get("installer_profile", {})) + return keys + + +def filter_compatible_recipes(root, directory, previous): + """Silently defer new engine capabilities, retaining compatible old apps.""" + from .catalog import Catalog + supported = supported_profile_keys(root) + candidate = Catalog(Path(root), catalog_dir=directory) + old = Catalog(Path(root), catalog_dir=previous) + old_items = {item["id"]: item for item in old.load_index()["applications"]} + index = json.loads((directory / "index.json").read_text()) + accepted = [] + old_fingerprints = None + for item in index["applications"]: + if item.get("hidden") or not item.get("automatic_install_candidate"): + accepted.append(item) + continue + template = candidate._compose_item(item) + profile = template.get("proxmox", {}).get("installer_profile", {}) + if set(profile) <= supported: + accepted.append(item) + continue + fallback = old_items.get(item["id"]) + if fallback is None: + continue + template = old._compose_item(fallback) + if not set(template.get("proxmox", {}).get("installer_profile", {})) <= supported: + continue + # Materialize the previously composed contract: its old overlay/volume + # policy must not be combined with a newly downloaded incompatible one. + app_id = item["id"] + for folder in ("apps", "curated"): + path = directory / folder / f"{app_id}.json" + path.parent.mkdir(exist_ok=True) + write_json(path, template) + (directory / "overlays" / f"{app_id}.json").unlink(missing_ok=True) + if old_fingerprints is None: + old_fingerprints = visible_recipes(old) + fallback = dict(fallback, _retained_recipe_hash=old_fingerprints.get(app_id), template_family="curated-profile", + curated_path=f"catalog/curated/{app_id}.json", + _retained_recipe=True) + accepted.append(fallback) + index["applications"] = accepted + write_json(directory / "index.json", index) + + def current_catalog(root, channel=None): root = Path(root) channel = channel or channel_for(root) @@ -102,14 +210,15 @@ def current_catalog(root, channel=None): revision = value.get("revision", "") if (isinstance(revision, str) and SHA.fullmatch(revision) and value.get("schema_hash") == schema_hash(root) - and value.get("engine_api") == CATALOG_API): + and value.get("engine_api") == CATALOG_API + and value.get("bundle_hash") == bundle_hash(root)): snapshot = value.get("snapshot", revision) if not isinstance(snapshot, str) or not re.fullmatch(revision + r"(?:-[0-9a-f]{12})?", snapshot): return root / "catalog" path = cache_root(root) / "snapshots" / snapshot index = read_json(path / "index.json") if (index.get("kind") == "proxmenux.oci-catalog-index" - and isinstance(index.get("applications"), list) and index["applications"]): + and isinstance(index.get("applications"), list)): return path return root / "catalog" @@ -146,7 +255,7 @@ def catalog_path(name): return path -def validate_snapshot(root, directory): +def validate_snapshot(root, directory, *, allow_empty=False): from .catalog import Catalog from jsonschema import Draft202012Validator compatibility = (json.loads((directory / "compatibility.json").read_text()) @@ -171,7 +280,7 @@ def validate_snapshot(root, directory): if not curated.startswith("catalog/"): raise ValueError("Invalid curated recipe path") catalog_path(curated.removeprefix("catalog/")) - if not ids: + if not ids and not allow_empty: raise ValueError("The remote catalog is empty") catalog = Catalog(Path(root), catalog_dir=directory) for item in catalog.load_index()["applications"]: @@ -187,6 +296,8 @@ def synchronize(root, channel, fetch=request, force=False): pointer = cache / f"{channel}-current.json" previous = read_json(pointer) active = current_catalog(root, channel) + capabilities = hashlib.sha256(json.dumps(sorted(supported_profile_keys(root))).encode()).hexdigest() + force = force or previous.get("profile_capabilities") != capabilities etag = previous.get("etag") if active != root / "catalog" and not force else None url = f"https://api.github.com/repos/{REPOSITORY}/commits?sha={channel}&path=oci/catalog&per_page=1" body, etag = fetch(url, etag) @@ -255,21 +366,26 @@ def synchronize(root, channel, fetch=request, force=False): finally: pool.shutdown(wait=True, cancel_futures=True) validate_snapshot(root, staging) + filter_compatible_recipes(root, staging, active) + validate_snapshot(root, staging, allow_empty=True) destination = (active if previous.get("revision") == revision and active != root / "catalog" else snapshots / revision) if destination.exists(): + staged_files = {str(p.relative_to(staging)) for p in staging.rglob("*.json")} intact = all((destination / relative).is_file() and not (destination / relative).is_symlink() - and git_blob((destination / relative).read_bytes()) == entry["sha"] - for relative, entry in entries.items()) + and (destination / relative).read_bytes() == (staging / relative).read_bytes() + for relative in staged_files) existing = {str(p.relative_to(destination)) for p in destination.rglob("*.json")} - if not intact or existing != set(entries): + if not intact or existing != staged_files: # Repair without removing a snapshot another wizard may still hold. destination = snapshots / (revision + "-" + uuid.uuid4().hex[:12]) - if not destination.exists(): - os.replace(staging, destination) - write_json(pointer, {"revision": revision, "snapshot": destination.name, "etag": etag, - "schema_hash": schema_hash(root), "engine_api": CATALOG_API}) + with snapshot_lock(root): + if not destination.exists(): + os.replace(staging, destination) + write_json(pointer, {"revision": revision, "snapshot": destination.name, "etag": etag, + "schema_hash": schema_hash(root), "engine_api": CATALOG_API, + "bundle_hash": bundle_hash(root), "profile_capabilities": capabilities}) return destination != active @@ -286,6 +402,8 @@ def worker(root, channel, force=False): write_json(cache / f"{channel}-result.json", {"ok": True, "changed": changed}) except Exception as error: write_json(cache / f"{channel}-result.json", {"ok": False, "error": str(error)[:400]}) + finally: + prune_snapshots(root) def visible_recipes(catalog): @@ -295,6 +413,9 @@ def visible_recipes(catalog): for item in catalog.load_index()["applications"]: if item.get("hidden") or not item.get("automatic_install_candidate"): continue + if item.get("_retained_recipe") and item.get("_retained_recipe_hash"): + result[item["id"]] = item["_retained_recipe_hash"] + continue directory = catalog.root / "catalog" if item["id"] in catalog.local_app_ids else catalog.catalog_dir if item.get("template_family") == "curated-profile": path = directory / item["curated_path"].removeprefix("catalog/") diff --git a/oci/tests/test_catalog_refresh.py b/oci/tests/test_catalog_refresh.py index cfdb32d3..e3e7b068 100644 --- a/oci/tests/test_catalog_refresh.py +++ b/oci/tests/test_catalog_refresh.py @@ -1,5 +1,6 @@ """Catalog delivery is atomic, offline-safe, and never acknowledges unseen apps.""" import json +import gc import errno import http.server import itertools @@ -74,6 +75,155 @@ class CatalogRefreshTests(unittest.TestCase): index["applications"].append(item) self.files["index.json"] = json.dumps(index).encode() + def future_profile(self, app_id): + index = json.loads(self.files["index.json"]) + item = next(x for x in index["applications"] if x["id"] == app_id) + path = item["curated_path"].removeprefix("catalog/") + recipe = json.loads(self.files[path]) + recipe.setdefault("proxmox", {}).setdefault("installer_profile", {})["future_engine_key"] = True + self.files[path] = json.dumps(recipe).encode() + + def test_new_unknown_profile_key_is_silent_and_not_counted(self): + self.sync() + baseline = Catalog(self.root) + status = refresh.CatalogRefresh(self.root, baseline) + self.revision = "b" * 40 + self.add_app() + self.future_profile("sample-added") + self.assertTrue(self.sync()) + catalog = Catalog(self.root) + self.assertNotIn("sample-added", refresh.visible_recipes(catalog)) + self.assertEqual(set(refresh.visible_recipes(catalog)), set(status.recipes)) + status.recipes = refresh.visible_recipes(catalog) + self.assertIn("New applications since your last visit: 0", status.describe()) + + def test_incompatible_existing_app_retains_complete_previous_contract(self): + self.sync() + old = Catalog(self.root).compose(self.item["id"]) + old_visible = refresh.visible_recipes(Catalog(self.root)) + self.revision = "b" * 40 + self.future_profile(self.item["id"]) + # A new policy must not mutate the retained contract. + policy = json.loads(self.files["volume-policy.json"]) + policy["shared_paths"] = [v["container_path"] for v in old["container_contract"].get("volumes", [])] + self.files["volume-policy.json"] = json.dumps(policy).encode() + self.sync() + self.assertEqual(Catalog(self.root).compose(self.item["id"]), old) + self.assertEqual(refresh.visible_recipes(Catalog(self.root)), old_visible) + # A later incompatible revision still retains the same original contract. + self.revision = "c" * 40 + self.sync() + self.assertEqual(Catalog(self.root).compose(self.item["id"]), old) + + def test_engine_capability_update_rechecks_same_revision_without_etag(self): + self.add_app() + self.future_profile("sample-added") + self.sync() + self.assertNotIn("sample-added", refresh.visible_recipes(Catalog(self.root))) + bundled = self.root / "catalog" / self.item["curated_path"].removeprefix("catalog/") + value = json.loads(bundled.read_text()) + value.setdefault("proxmox", {}).setdefault("installer_profile", {})["future_engine_key"] = True + bundled.write_text(json.dumps(value)) + self.requests.clear() + self.assertTrue(self.sync()) + self.assertIsNone(self.requests[0][1]) + self.assertIn("sample-added", refresh.visible_recipes(Catalog(self.root))) + + def test_known_profile_key_accepts_new_values(self): + self.add_app() + path = "curated/sample-added.json" + value = json.loads(self.files[path]) + self.assertIn("completion_notes", refresh.supported_profile_keys(self.root)) + value["proxmox"]["installer_profile"]["completion_notes"].append("A new supported setting") + self.files[path] = json.dumps(value).encode() + self.sync() + self.assertIn("sample-added", refresh.visible_recipes(Catalog(self.root))) + + def test_all_incompatible_apps_can_produce_empty_visible_catalog(self): + self.future_profile(self.item["id"]) + # No compatible old recipe exists on this host. + path = self.root / "catalog" / self.item["curated_path"].removeprefix("catalog/") + value = json.loads(path.read_text()) + value["proxmox"]["installer_profile"].pop("future_engine_key", None) + # Use a new ID instead of retaining the bundled application. + self.add_app() + self.future_profile("sample-added") + index = json.loads(self.files["index.json"]) + index["applications"] = [x for x in index["applications"] if x["id"] == "sample-added"] + self.files["index.json"] = json.dumps(index).encode() + self.sync() + self.assertNotEqual(refresh.current_catalog(self.root), self.root / "catalog") + self.assertEqual(refresh.visible_recipes(Catalog(self.root)), {}) + + def test_pruning_preserves_active_and_leased_snapshots_then_reclaims_reader(self): + self.sync() + old = Catalog(self.root) + old_directory = old.catalog_dir + self.revision = "b" * 40 + self.add_app() + self.sync() + refresh.prune_snapshots(self.root) + self.assertTrue(old_directory.exists()) + self.assertEqual(old.compose(self.item["id"])["id"], self.item["template_id"]) + del old + gc.collect() + refresh.prune_snapshots(self.root) + self.assertFalse(old_directory.exists()) + self.assertTrue(refresh.current_catalog(self.root).exists()) + + def test_pruning_preserves_both_channels_and_unlinks_orphan_lease(self): + self.sync() + main = refresh.current_catalog(self.root, "main") + self.revision = "b" * 40 + refresh.synchronize(self.root, "develop", self.fetch) + orphan = refresh.cache_root(self.root) / "snapshots" / ("c" * 40) + orphan.mkdir() + refresh.prune_snapshots(self.root) + self.assertTrue(main.exists()) + self.assertTrue(refresh.current_catalog(self.root, "develop").exists()) + self.assertFalse(orphan.exists()) + self.assertFalse((refresh.cache_root(self.root) / "leases" / orphan.name).exists()) + + def test_worker_prunes_even_when_no_change_or_network_failure(self): + self.sync() + for error in (None, OSError("offline")): + orphan = refresh.cache_root(self.root) / "snapshots" / ("c" * 40) + orphan.mkdir() + with patch.object(refresh, "synchronize", side_effect=error, return_value=False): + refresh.worker(self.root, "main") + self.assertFalse(orphan.exists()) + + def test_reader_process_exit_releases_lease_without_pid_cleanup(self): + import subprocess + self.sync() + directory = refresh.current_catalog(self.root) + code = ("import fcntl,sys,time; " + "f=open(sys.argv[1], 'a'); fcntl.flock(f,fcntl.LOCK_SH); " + "print('ready',flush=True); time.sleep(30)") + leases = refresh.cache_root(self.root) / "leases" + leases.mkdir(exist_ok=True) + child = subprocess.Popen([sys.executable, "-c", code, str(leases / directory.name)], + stdout=subprocess.PIPE, text=True) + self.addCleanup(child.stdout.close) + self.addCleanup(lambda: child.poll() is None and child.kill()) + self.assertEqual(child.stdout.readline().strip(), "ready") + self.revision = "b" * 40 + self.sync() + refresh.prune_snapshots(self.root) + self.assertTrue(directory.exists()) + child.kill() + child.wait(timeout=3) + refresh.prune_snapshots(self.root) + self.assertFalse(directory.exists()) + + def test_menu_pin_does_not_wait_for_network_worker_lock(self): + self.sync() + lock = (refresh.cache_root(self.root) / "refresh.lock").open("a") + self.addCleanup(lock.close) + refresh.fcntl.flock(lock, refresh.fcntl.LOCK_EX) + catalog = Catalog(self.root) + self.assertEqual(catalog.compose(self.item["id"])["id"], self.item["template_id"]) + def test_complete_catalog_is_selected_atomically_without_redownloading_unchanged_files(self): self.assertTrue(self.sync()) directory = refresh.current_catalog(self.root, "main")