Prune catalog snapshots and defer recipes requiring newer profile keys

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

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"],

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/")

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")