diff --git a/oci/CATALOG-REFRESH.md b/oci/CATALOG-REFRESH.md index 17aa792e..f87a627b 100644 --- a/oci/CATALOG-REFRESH.md +++ b/oci/CATALOG-REFRESH.md @@ -6,6 +6,10 @@ catalog, and changed recipes. Returning from a search or installation does not start another download. A running wizard keeps the catalog it started with. The menu also offers **Refresh catalog** to check again manually or retry after a connection failure. Selecting it during a refresh does not start a second worker. +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. The worker checks the latest commit affecting `oci/catalog` in the official MacRimi/ProxMenux repository. GitHub's ETag avoids repeated transfers. When the @@ -20,6 +24,8 @@ the engine replaced by the normal ProxMenux updater. The bundled catalog remains the fallback. A network failure, incomplete manifest, invalid recipe, different template schema or unsupported catalog engine API leaves the previous catalog active. The private `CHANNEL-result.json` records the diagnostic. +The revision, snapshot and ETag are committed together in one atomic pointer; +failure to publish it cannot acknowledge an update that was not applied. The channel follows `install_branch`, an active beta program or `beta_version.txt`; otherwise it is `main`. A host can explicitly select `main` or `develop` with diff --git a/oci/src/proxmenux_oci/catalog_refresh.py b/oci/src/proxmenux_oci/catalog_refresh.py index f733f038..d78a95fc 100644 --- a/oci/src/proxmenux_oci/catalog_refresh.py +++ b/oci/src/proxmenux_oci/catalog_refresh.py @@ -14,12 +14,15 @@ import json import os from pathlib import Path, PurePosixPath import re +import shutil import subprocess import sys import tempfile import threading +import time import urllib.error import urllib.request +import uuid from .i18n import translate @@ -46,11 +49,12 @@ def channel_for(root): root.parent if root.name == "oci" else root) for name in ("oci/catalog-source.json", "config.json"): try: - value = json.loads((base / name).read_text()) + value = read_json(base / name) channel = value.get("channel") or value.get("install_branch") if channel in ("main", "develop"): return channel - if (value.get("beta_program") or {}).get("status") == "active": + beta = value.get("beta_program") + if isinstance(beta, dict) and beta.get("status") == "active": return "develop" except (OSError, ValueError, TypeError): pass @@ -99,7 +103,10 @@ def current_catalog(root, channel=None): if (isinstance(revision, str) and SHA.fullmatch(revision) and value.get("schema_hash") == schema_hash(root) and value.get("engine_api") == CATALOG_API): - path = cache_root(root) / "snapshots" / revision + 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"]): @@ -122,6 +129,7 @@ def request(url, etag=None, limit=MAX_FILE): raise ValueError("Catalog response exceeds the size limit") return body, response.headers.get("ETag") except urllib.error.HTTPError as error: + error.close() if error.code == 304: return None, etag raise @@ -171,15 +179,15 @@ def validate_snapshot(root, directory): validator.validate(catalog._compose_item(item)) -def synchronize(root, channel, fetch=request): +def synchronize(root, channel, fetch=request, force=False): """Download changed JSON files into a private snapshot, then publish once.""" root = Path(root) cache = cache_root(root) cache.mkdir(parents=True, exist_ok=True) - metadata_path = cache / f"{channel}-remote.json" - previous = read_json(metadata_path) + pointer = cache / f"{channel}-current.json" + previous = read_json(pointer) active = current_catalog(root, channel) - etag = previous.get("etag") if active != root / "catalog" else None + 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) if body is None: @@ -188,8 +196,8 @@ def synchronize(root, channel, fetch=request): revision = commits[0]["sha"] if not isinstance(revision, str) or not SHA.fullmatch(revision): raise ValueError("Invalid catalog revision") - if active.name == revision: - write_json(metadata_path, {"etag": etag}) + if previous.get("revision") == revision and active != root / "catalog" and not force: + write_json(pointer, {**previous, "etag": etag}) return False body, _ = fetch(f"https://api.github.com/repos/{REPOSITORY}/git/trees/{revision}?recursive=1") tree = json.loads(body) @@ -211,6 +219,8 @@ def synchronize(root, channel, fetch=request): raise ValueError("Invalid catalog manifest entry") if type(entry.get("size")) is not int or not 0 <= entry["size"] <= MAX_FILE: raise ValueError("Invalid catalog file size") + if relative in entries: + raise ValueError("Duplicate catalog manifest path") entries[relative] = entry if schema != git_blob((root / "schemas/oci-template.schema.json").read_bytes()): raise ValueError("Catalog schema changed; update ProxMenux first") @@ -222,8 +232,11 @@ def synchronize(root, channel, fetch=request): snapshots.mkdir(exist_ok=True) with tempfile.TemporaryDirectory(prefix=".download-", dir=snapshots) as name: staging = Path(name) + deadline = time.monotonic() + 90 def download(pair): + if time.monotonic() > deadline: + raise TimeoutError("Catalog download exceeded its time limit") relative, entry = pair old = active / relative data = old.read_bytes() if old.is_file() else b"" @@ -236,28 +249,40 @@ def synchronize(root, channel, fetch=request): target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(data) - with ThreadPoolExecutor(max_workers=4) as pool: + pool = ThreadPoolExecutor(max_workers=4) + try: list(pool.map(download, entries.items())) + finally: + pool.shutdown(wait=True, cancel_futures=True) validate_snapshot(root, staging) - destination = snapshots / revision + destination = (active if previous.get("revision") == revision and active != root / "catalog" + else snapshots / revision) if destination.exists(): - # Revalidate an existing immutable snapshot after an interrupted publication. - validate_snapshot(root, destination) - else: + 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()) + existing = {str(p.relative_to(destination)) for p in destination.rglob("*.json")} + if not intact or existing != set(entries): + # 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(cache / f"{channel}-current.json", {"revision": revision, + write_json(pointer, {"revision": revision, "snapshot": destination.name, "etag": etag, "schema_hash": schema_hash(root), "engine_api": CATALOG_API}) - write_json(metadata_path, {"etag": etag}) - return True + return destination != active -def worker(root, channel): +def worker(root, channel, force=False): cache = cache_root(root) cache.mkdir(parents=True, exist_ok=True) with (cache / "refresh.lock").open("a") as lock: fcntl.flock(lock, fcntl.LOCK_EX) try: - changed = synchronize(root, channel) + for partial in (cache / "snapshots").glob(".download-*"): + if partial.is_dir() and not partial.is_symlink(): + shutil.rmtree(partial) + changed = synchronize(root, channel, force=force) 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]}) @@ -300,6 +325,7 @@ class CatalogRefresh: self.channel = channel_for(root) self.seen_path = cache_root(root) / f"{self.channel}-seen.json" self.recipes = visible_recipes(catalog) + self.displayed_directory = catalog.catalog_dir baseline = read_json(self.seen_path).get("recipes") self.baseline = baseline if isinstance(baseline, dict) else self.recipes.copy() self.process = None @@ -310,14 +336,17 @@ class CatalogRefresh: self.start() return self - def start(self): + def start(self, force=False): if self.active: return False self._finished = False self.failed = False environment = dict(os.environ, PYTHONPATH=str(self.root / "src")) try: - self.process = subprocess.Popen([sys.executable, "-m", __name__, str(self.root), self.channel], + command = [sys.executable, "-m", __name__, str(self.root), self.channel] + if force: + command.append("--force") + self.process = subprocess.Popen(command, env=environment, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True) @@ -348,7 +377,11 @@ class CatalogRefresh: self._finished = True result = read_json(cache_root(self.root) / f"{self.channel}-result.json") self.failed = self.process.returncode != 0 or not result.get("ok") - return not self.failed and bool(result.get("changed")) + directory = current_catalog(self.root, self.channel) + changed = directory != self.displayed_directory + if changed: + self.displayed_directory = directory + return changed def describe(self): text = translate("Available applications: {count}").format(count=len(self.recipes)) @@ -374,5 +407,6 @@ if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("root", type=Path) parser.add_argument("channel", choices=("main", "develop")) + parser.add_argument("--force", action="store_true") args = parser.parse_args() - worker(args.root, args.channel) + worker(args.root, args.channel, args.force) diff --git a/oci/src/proxmenux_oci/cli.py b/oci/src/proxmenux_oci/cli.py index 47407ec0..e2c9025f 100644 --- a/oci/src/proxmenux_oci/cli.py +++ b/oci/src/proxmenux_oci/cli.py @@ -626,7 +626,7 @@ def interactive(catalog: Catalog) -> int: from .management import interactive_management interactive_management(PROJECT_ROOT, ui) elif selection == "catalog-refresh": - catalog_refresh.start() + catalog_refresh.start(force=True) elif selection == "updates": if not updates.active: if state['available']: diff --git a/oci/tests/test_catalog_refresh.py b/oci/tests/test_catalog_refresh.py index f2f8c449..cfdb32d3 100644 --- a/oci/tests/test_catalog_refresh.py +++ b/oci/tests/test_catalog_refresh.py @@ -1,10 +1,17 @@ """Catalog delivery is atomic, offline-safe, and never acknowledges unseen apps.""" import json +import errno +import http.server +import itertools from pathlib import Path import shutil import sys import tempfile +import threading +import time import unittest +import urllib.error +import urllib.parse from unittest.mock import Mock, patch ROOT = Path(__file__).resolve().parents[1] @@ -303,6 +310,272 @@ class CatalogRefreshTests(unittest.TestCase): options = dict(_catalog_options(Catalog(self.root))[-1]) self.assertEqual(options['catalog-refresh'], 'Refresh catalog') + def test_sequence_new_app_failed_update_retry_and_withdrawal(self): + self.sync() + old = Catalog(self.root) + status = refresh.CatalogRefresh(self.root, old) + status.mark_seen() + self.revision = 'b' * 40 + self.add_app() + self.assertTrue(self.sync()) + self.assertIn('sample-added', refresh.visible_recipes(Catalog(self.root))) + self.assertNotIn('sample-added', refresh.visible_recipes(old)) # Wizard remains pinned. + next_visit = refresh.CatalogRefresh(self.root, Catalog(self.root)) + self.assertIn('New applications since your last visit: 1', next_visit.describe()) + next_visit.mark_seen() + self.revision = 'c' * 40 + recipe_path = self.item['curated_path'].removeprefix('catalog/') + good = self.files[recipe_path] + self.files[recipe_path] = b'{invalid-json' + with self.assertRaises(ValueError): + self.sync() + self.assertEqual(refresh.current_catalog(self.root).name, 'b' * 40) + self.files[recipe_path] = good + self.assertTrue(self.sync()) + index = json.loads(self.files['index.json']) + index['applications'] = [self.item] + self.files['index.json'] = json.dumps(index).encode() + for name in ('apps/sample-added.json', 'curated/sample-added.json'): + self.files.pop(name) + self.revision = 'd' * 40 + self.assertTrue(self.sync()) + current = Catalog(self.root) + self.assertNotIn('sample-added', refresh.visible_recipes(current)) + self.assertFalse((current.catalog_dir / 'curated/sample-added.json').exists()) + self.assertIn('New applications since your last visit: 0', + refresh.CatalogRefresh(self.root, current).describe()) + + def test_http_errors_and_timeouts_keep_snapshot_then_retry_succeeds(self): + self.sync() + old = refresh.current_catalog(self.root) + errors = [TimeoutError('timeout'), urllib.error.URLError('DNS unavailable')] + errors += [urllib.error.HTTPError('https://api.github.com', code, 'test', {}, None) + for code in (403, 404, 429, 500, 503)] + for error in errors: + if isinstance(error, urllib.error.HTTPError): self.addCleanup(error.close) + self.revision = 'b' * 40 + for error in errors: + with self.subTest(error=str(error)), self.assertRaises(Exception): + refresh.synchronize(self.root, 'main', Mock(side_effect=error)) + self.assertEqual(refresh.current_catalog(self.root), old) + self.assertTrue(self.sync()) + + def test_failed_file_transfer_does_not_publish_partial_download(self): + self.sync(); old = refresh.current_catalog(self.root) + self.revision = 'b' * 40; self.add_app() + def interrupted(url, etag=None): + if 'raw.githubusercontent' in url: + raise TimeoutError('Connection lost during download') + return self.fetch(url, etag) + with self.assertRaises(TimeoutError): + refresh.synchronize(self.root, 'main', interrupted) + self.assertEqual(refresh.current_catalog(self.root), old) + self.assertFalse(list((refresh.cache_root(self.root) / 'snapshots').glob('.download-*'))) + self.assertTrue(self.sync()) + + def test_disk_full_and_pointer_permission_failure_preserve_previous_revision_and_etag(self): + for error in (OSError(errno.ENOSPC, 'Disk full'), PermissionError('Read only')): + with self.subTest(error=error): + self.sync() + pointer = refresh.cache_root(self.root) / 'main-current.json' + before = pointer.read_bytes() + old = refresh.current_catalog(self.root) + self.revision = 'b' * 40 + real = refresh.write_json + def fail(path, value): + if Path(path) == pointer: + raise error + return real(path, value) + with patch.object(refresh, 'write_json', side_effect=fail), self.assertRaises(OSError): + self.sync() + self.assertEqual(pointer.read_bytes(), before) + self.assertEqual(refresh.current_catalog(self.root), old) + self.assertTrue(self.sync()) + self.revision = 'c' * 40 + + def test_manual_force_repairs_corrupted_recipe_at_same_revision(self): + self.sync() + old = refresh.current_catalog(self.root) + recipe = old / self.item['curated_path'].removeprefix('catalog/') + recipe.write_text('{}') + self.requests.clear() + self.assertTrue(refresh.synchronize(self.root, 'main', self.fetch, force=True)) + self.assertIsNone(self.requests[0][1]) + self.assertNotEqual(refresh.current_catalog(self.root), old) + self.assertEqual(Catalog(self.root).compose(self.item['id'])['id'], self.item['template_id']) + self.assertFalse(refresh.synchronize(self.root, 'main', self.fetch, force=True)) + + def test_corrupt_cached_index_can_be_repaired_by_automatic_refresh(self): + self.sync() + old = refresh.current_catalog(self.root) + (old / 'index.json').write_text('{broken') + self.assertTrue(self.sync()) + self.assertNotEqual(refresh.current_catalog(self.root), self.root / 'catalog') + self.assertEqual(Catalog(self.root).compose(self.item['id'])['id'], self.item['template_id']) + + def test_second_session_result_cannot_hide_successful_update(self): + status = refresh.CatalogRefresh(self.root, Catalog(self.root)) + self.sync() + status.process = Mock(returncode=0) + status.process.poll.return_value = 0 + # Another session checked the newly installed revision and found no change. + refresh.write_json(refresh.cache_root(self.root) / 'main-result.json', + {'ok': True, 'changed': False}) + self.assertTrue(status.consume()) + + def test_crashed_worker_does_not_trust_stale_success(self): + status = refresh.CatalogRefresh(self.root, Catalog(self.root)) + status.process = Mock(returncode=1); status.process.poll.return_value = 1 + refresh.write_json(refresh.cache_root(self.root) / 'main-result.json', + {'ok': True, 'changed': True}) + self.assertFalse(status.consume()) + self.assertTrue(status.failed) + + def test_force_manual_refresh_passes_flag_to_worker(self): + status = refresh.CatalogRefresh(self.root, Catalog(self.root)) + with patch.object(refresh.subprocess, 'Popen') as spawn: + self.assertTrue(status.start(force=True)) + self.assertIn('--force', spawn.call_args.args[0]) + + def test_duplicate_manifest_path_symlink_and_truncated_tree_are_rejected(self): + for change in ('duplicate', 'symlink', 'truncated'): + def bad_tree(url, etag=None): + body, tag = self.fetch(url, etag) + if '/git/trees/' in url: + tree = json.loads(body) + if change == 'duplicate': tree['tree'].append(tree['tree'][0]) + elif change == 'symlink': tree['tree'][0]['mode'] = '120000' + else: tree['truncated'] = True + body = json.dumps(tree).encode() + return body, tag + with self.subTest(change=change), self.assertRaises(ValueError): + refresh.synchronize(self.root, 'main', bad_tree) + self.assertEqual(refresh.current_catalog(self.root), self.root / 'catalog') + + def test_bad_channel_config_falls_back_without_crashing_menu(self): + config = self.root.parent / 'catalog-source.json' + for value in ([], None, 'invalid', {'channel': 'other'}, {'beta_program': 'invalid'}): + config.write_text(json.dumps(value)) + self.assertEqual(refresh.channel_for(self.root), 'main') + + def test_full_disk_during_file_write_leaves_no_partial_catalog(self): + self.sync(); old = refresh.current_catalog(self.root) + self.revision = 'b' * 40; self.add_app() + real = Path.write_bytes + def full(path, data): + if any(part.startswith('.download-') for part in path.parts): + raise OSError(errno.ENOSPC, 'Disk full') + return real(path, data) + with patch.object(Path, 'write_bytes', new=full), self.assertRaises(OSError): + self.sync() + self.assertEqual(refresh.current_catalog(self.root), old) + self.assertFalse(list((refresh.cache_root(self.root) / 'snapshots').glob('.download-*'))) + + def test_orphaned_download_is_removed_before_retry(self): + partial = refresh.cache_root(self.root) / 'snapshots/.download-crashed' + partial.mkdir(parents=True); (partial/'partial.json').write_text('{}') + synchronize = refresh.synchronize + with patch.object(refresh, 'synchronize', side_effect=lambda *a, **kw: + synchronize(self.root, 'main', self.fetch)): + refresh.worker(self.root, 'main') + self.assertFalse(partial.exists()) + self.assertTrue(refresh.read_json(refresh.cache_root(self.root)/'main-result.json')['ok']) + + def test_request_rejects_oversized_response_and_handles_304(self): + context = Mock() + context.__enter__ = Mock(return_value=Mock(read=Mock(return_value=b'oversized'))) + context.__exit__ = Mock(return_value=False) + with patch.object(refresh.urllib.request, 'urlopen', return_value=context), self.assertRaises(ValueError): + refresh.request('https://api.github.com/test', limit=2) + error = urllib.error.HTTPError('https://api.github.com', 304, 'unchanged', {}, None) + with patch.object(refresh.urllib.request, 'urlopen', side_effect=error): + self.assertEqual(refresh.request('https://api.github.com/test', etag='old'), (None, 'old')) + + def test_unwritable_seen_state_does_not_crash_menu(self): + status = refresh.CatalogRefresh(self.root, Catalog(self.root)) + with patch.object(refresh, 'write_json', side_effect=PermissionError('Read only')): + status.mark_seen() + self.assertTrue(status.failed) + + def test_concurrent_workers_serialize_the_catalog_transaction(self): + barrier = threading.Barrier(2) + guard = threading.Lock() + active = maximum = 0 + errors = [] + def operation(*args, **kwargs): + nonlocal active, maximum + with guard: + active += 1; maximum = max(maximum, active) + time.sleep(0.03) + with guard: active -= 1 + return False + def run(): + try: + barrier.wait(timeout=2) + refresh.worker(self.root, 'main') + except Exception as error: + errors.append(error) + with patch.object(refresh, 'synchronize', side_effect=operation): + threads = [threading.Thread(target=run) for _ in range(2)] + for thread in threads: thread.start() + for thread in threads: thread.join(timeout=3) + self.assertFalse(errors) + self.assertTrue(all(not thread.is_alive() for thread in threads)) + self.assertEqual(maximum, 1) + + def test_download_deadline_preserves_previous_catalog(self): + self.sync(); old = refresh.current_catalog(self.root) + self.revision = 'b' * 40; self.add_app() + with patch.object(refresh.time, 'monotonic', side_effect=itertools.chain([0], itertools.repeat(100))), \ + self.assertRaises(TimeoutError): + self.sync() + self.assertEqual(refresh.current_catalog(self.root), old) + + def test_http_end_to_end_new_release_outage_retry_manual_refresh_and_repair(self): + owner = self + unavailable = False + class Handler(http.server.BaseHTTPRequestHandler): + def log_message(self, *args): pass + def do_GET(self): + if unavailable: + self.send_error(503); return + etag = '"' + owner.revision + '"' + if self.headers.get('If-None-Match') == etag: + self.send_response(304); self.end_headers(); return + host = ('https://api.github.com' if self.path.startswith('/repos/') + else 'https://raw.githubusercontent.com') + body, _ = owner.fetch(host + self.path) + self.send_response(200) + self.send_header('Content-Length', str(len(body))) + self.send_header('ETag', etag) + self.end_headers(); self.wfile.write(body) + server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + def fetch(url, etag=None): + original = urllib.parse.urlsplit(url) + path = original.path + ('?' + original.query if original.query else '') + return refresh.request(f'http://127.0.0.1:{server.server_port}' + path, etag) + try: + self.assertTrue(refresh.synchronize(self.root, 'main', fetch)) + self.assertFalse(refresh.synchronize(self.root, 'main', fetch)) + first = refresh.current_catalog(self.root) + unavailable = True + self.revision = 'b' * 40; self.add_app() + with self.assertRaises(urllib.error.HTTPError): + refresh.synchronize(self.root, 'main', fetch) + self.assertEqual(refresh.current_catalog(self.root), first) + unavailable = False + self.assertTrue(refresh.synchronize(self.root, 'main', fetch)) + current = Catalog(self.root) + self.assertEqual(len(refresh.visible_recipes(current)), 2) + self.assertFalse(refresh.synchronize(self.root, 'main', fetch, force=True)) + (current.catalog_dir / 'curated/sample-added.json').write_text('{}') + self.assertTrue(refresh.synchronize(self.root, 'main', fetch, force=True)) + self.assertEqual(Catalog(self.root).compose('sample-added')['id'], 'image-sample-added') + finally: + server.shutdown(); server.server_close(); thread.join(timeout=2) + if __name__ == "__main__": unittest.main()