diff --git a/AppImage/scripts/native_app_discovery.py b/AppImage/scripts/native_app_discovery.py index 29a56b7b..ac9daa9e 100644 --- a/AppImage/scripts/native_app_discovery.py +++ b/AppImage/scripts/native_app_discovery.py @@ -17,6 +17,7 @@ """Separate from catalog detection: callers must explicitly request a scan.""" import copy +from contextlib import contextmanager import fcntl import hashlib import json @@ -25,6 +26,7 @@ from pathlib import Path import re import selectors import signal +import stat import subprocess import threading import time @@ -40,6 +42,7 @@ OCI_ROOT = Path('/usr/local/share/proxmenux/oci/instances') UPDATE_LOCK_ROOT = Path('/run/lock') _cache = {} _locks = {} +_generations = {} _mutex = threading.Lock() _PATH = re.compile(r'/[A-Za-z0-9_.+/@-]{1,510}\Z') _NAME = re.compile(r'[A-Za-z0-9][A-Za-z0-9_.+@-]{0,63}\Z') @@ -509,7 +512,13 @@ def _guard(vmid, deadline, run): raise ProbeError('guest_busy') update_lock = UPDATE_LOCK_ROOT / f'proxmenux-lxc-update-{vmid}.lock' try: - with update_lock.open('rb') as handle: + fd = os.open(update_lock, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + with os.fdopen(fd, 'rb') as handle: + info = os.fstat(handle.fileno()) + if not stat.S_ISREG(info.st_mode): + raise ProbeError('lock_unavailable') + if info.st_size: + raise ProbeError('guest_busy') try: fcntl.flock(handle, fcntl.LOCK_SH | fcntl.LOCK_NB) except BlockingIOError: @@ -529,9 +538,45 @@ def _guard(vmid, deadline, run): raise ProbeError('guest_stopped' if status == 'status: stopped' else 'status_unknown') +@contextmanager +def _scan_reservation(vmid): + """Exclude cooperating updates/control tasks for this bounded scan only. + + Keep an empty lock inode after first use, as the update runner does. + Never truncate a pending control-task reservation or unlink its inode. + """ + path = UPDATE_LOCK_ROOT / f'proxmenux-lxc-update-{vmid}.lock' + fd = None + try: + fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600) + if not stat.S_ISREG(os.fstat(fd).st_mode): + raise ProbeError('lock_unavailable') + try: + fcntl.flock(fd, fcntl.LOCK_SH | fcntl.LOCK_NB) + except BlockingIOError: + raise ProbeError('guest_busy') from None + if os.fstat(fd).st_size: + raise ProbeError('guest_busy') + yield + except OSError: + raise ProbeError('lock_unavailable') from None + finally: + if fd is not None: + os.close(fd) + + def scan(vmid, run=bounded_run): deadline = time.monotonic() + TOTAL_SECONDS _guard(vmid, deadline, run) + with _scan_reservation(vmid): + result = _scan_metadata(vmid, deadline, run) + # External PVE lifecycle operations do not share our flock. Fail closed + # if the guest stopped, changed ownership or became locked meanwhile. + _guard(vmid, deadline, run) + return result + + +def _scan_metadata(vmid, deadline, run): records, issues, consumed, processes, listeners, hestia_panels = [], [], 0, [], [], [] for kind, script in (('S', SERVICE_SCRIPT), ('P', PROCESS_SCRIPT), ('L', PORT_SCRIPT)): try: @@ -637,6 +682,7 @@ def cached(vmid): def mark_stale(vmid): """Lifecycle events retire freshness, never collect new guest metadata.""" with _mutex: + _generations[int(vmid)] = _generations.get(int(vmid), 0) + 1 if int(vmid) in _cache: _cache[int(vmid)]['stale'] = True @@ -648,23 +694,28 @@ def discover(vmid): if not lock.acquire(blocking=False): return {**cached(vmid), 'status': 'busy', 'stale': True} try: + with _mutex: + generation = _generations.get(vmid, 0) try: result = scan(vmid) except ProbeError as exc: result = {**cached(vmid), 'status': 'error', 'issues': [str(exc)], 'stale': True} result.setdefault('scanned_at', time.time()) - with _mutex: - _cache[vmid] = copy.deepcopy(result) - return result - previous = cached(vmid) - if result['status'] != 'complete' and previous['candidates']: - # Partial output cannot prove that previously seen apps disappeared. - ids = {item['discovery_id'] for item in result['candidates']} - result['candidates'].extend(item for item in previous['candidates'] - if item['discovery_id'] not in ids) - result['candidates'] = result['candidates'][:MAX_CANDIDATES] - result['stale'] = True + # Comparing the generation and publishing must be atomic with lifecycle + # invalidation, including the first scan when no cache exists yet. with _mutex: + previous = copy.deepcopy(_cache.get(vmid)) + if _generations.get(vmid, 0) != generation: + result = {**(previous or {'vmid': vmid, 'candidates': [], + 'scanned_at': time.time()}), + 'status': 'error', 'issues': ['guest_changed'], 'stale': True} + elif result['status'] != 'complete' and previous and previous['candidates']: + # Partial output cannot prove that previously seen apps disappeared. + ids = {item['discovery_id'] for item in result['candidates']} + result['candidates'].extend(item for item in previous['candidates'] + if item['discovery_id'] not in ids) + result['candidates'] = result['candidates'][:MAX_CANDIDATES] + result['stale'] = True _cache[vmid] = copy.deepcopy(result) return result finally: diff --git a/AppImage/scripts/tests/test_native_app_discovery.py b/AppImage/scripts/tests/test_native_app_discovery.py index 5e90b4cf..397bb3fc 100644 --- a/AppImage/scripts/tests/test_native_app_discovery.py +++ b/AppImage/scripts/tests/test_native_app_discovery.py @@ -42,6 +42,7 @@ class DiscoveryFixtures(unittest.TestCase): self.addCleanup(context.stop) native._cache.clear() native._locks.clear() + native._generations.clear() self.calls = [] def runner(self, services=None, processes='END\n', footprint='FILE\n', manifest=None, @@ -375,6 +376,142 @@ class DiscoveryFixtures(unittest.TestCase): with patch.object(native, 'scan', return_value={**complete, 'candidates': []}): self.assertEqual(native.discover(101)['candidates'], []) + def test_lifecycle_during_first_or_cached_scan_cannot_publish_fresh_result(self): + complete = native.scan(101, self.runner()) + for prior in (None, complete): + with self.subTest(cached=prior is not None): + native._cache.clear() + if prior: + native._cache[101] = copy.deepcopy(prior) + entered, release = threading.Event(), threading.Event() + results = [] + def delayed(vmid): + entered.set() + if not release.wait(3): + raise AssertionError('scan release timed out') + return copy.deepcopy(complete) + with patch.object(native, 'scan', side_effect=delayed): + worker = threading.Thread(target=lambda: results.append(native.discover(101))) + worker.start() + try: + self.assertTrue(entered.wait(3)) + native.mark_stale(101) + finally: + release.set() + worker.join(3) + self.assertFalse(worker.is_alive()) + self.assertEqual(len(results), 1) + self.assertTrue(results[0]['stale']) + self.assertIn('guest_changed', results[0]['issues']) + self.assertEqual(results[0]['candidates'], prior['candidates'] if prior else []) + self.assertTrue(native.cached(101)['stale']) + with patch.object(native, 'scan', return_value=copy.deepcopy(complete)): + self.assertFalse(native.discover(101)['stale']) + with patch.object(native, 'scan', side_effect=lambda vmid: ( + native.mark_stale(102), copy.deepcopy(complete))[1]): + self.assertFalse(native.discover(101)['stale']) + + def test_scan_excludes_later_updater_and_releases_lock_after_success_or_failure(self): + path = self.root / 'proxmenux-lxc-update-101.lock' + for fail in (False, True): + with self.subTest(fail=fail): + runner = self.runner() + probes = [] + def checked(argv, deadline): + if argv[1] == 'exec': + attempt = subprocess.run(['flock', '-n', str(path), 'true']) + self.assertEqual(attempt.returncode, 1) + probes.append(argv) + if fail: + raise native.ProbeError('timeout') + return runner(argv, deadline) + native.scan(101, checked) + self.assertTrue(probes) + self.assertEqual(subprocess.run(['flock', '-n', str(path), 'true']).returncode, 0) + self.assertEqual(path.read_bytes(), b'') + + def test_pending_control_reservation_blocks_scan_without_touching_it(self): + path = self.root / 'proxmenux-lxc-update-101.lock' + original = b'{"task":"pending-control"}' + path.write_bytes(original) + with self.assertRaisesRegex(native.ProbeError, 'guest_busy'): + native.scan(101, self.runner()) + self.assertFalse(any(call[1] == 'exec' for call in self.calls)) + self.assertEqual(path.read_bytes(), original) + + def test_guest_stop_during_scan_never_returns_complete(self): + runner = self.runner() + statuses = iter(['status: running', 'status: stopped']) + def stopped(argv, deadline): + return next(statuses) if argv[1] == 'status' else runner(argv, deadline) + with self.assertRaisesRegex(native.ProbeError, 'guest_stopped'): + native.scan(101, stopped) + self.assertTrue(all(call[1] == 'exec' for call in self.calls)) + + def test_scan_reservation_closes_race_after_guard_and_releases_after_exception(self): + import lxc_check_job + path = self.root / 'proxmenux-lxc-update-101.lock' + runner = self.runner() + handles = [] + def starts_update(argv, deadline): + if argv[1] == 'status': + handles.append(lxc_check_job.acquire(101)) + return runner(argv, deadline) + with patch.object(lxc_check_job, 'LOCK_DIR', self.root): + try: + with self.assertRaisesRegex(native.ProbeError, 'guest_busy'): + native.scan(101, starts_update) + self.assertFalse(any(call[1] == 'exec' for call in self.calls)) + finally: + for handle in handles: + handle.close() + def crash(argv, deadline): + if argv[1] == 'exec': + with self.assertRaises(BlockingIOError): + lxc_check_job.acquire(101) + # Another CT must remain available. + lxc_check_job.acquire(102).close() + raise RuntimeError('unexpected probe failure') + return runner(argv, deadline) + with self.assertRaisesRegex(RuntimeError, 'unexpected'): + native.scan(101, crash) + lxc_check_job.acquire(101).close() + self.assertEqual(path.read_bytes(), b'') + + def test_failed_scan_releases_discovery_lock_and_keeps_prior_evidence(self): + complete = native.scan(101, self.runner()) + native._cache[101] = copy.deepcopy(complete) + with patch.object(native, 'scan', side_effect=RuntimeError('unexpected')): + with self.assertRaises(RuntimeError): + native.discover(101) + self.assertFalse(native._locks[101].locked()) + self.assertEqual(native.cached(101)['candidates'], complete['candidates']) + + def test_invalid_lock_inode_and_unreadable_config_fail_before_any_probe(self): + import os + path = self.root / 'proxmenux-lxc-update-101.lock' + target = self.root / 'unrelated' + target.write_bytes(b'keep') + for kind in ('symlink', 'fifo', 'directory'): + with self.subTest(kind=kind): + if kind == 'symlink': + path.symlink_to(target) + elif kind == 'fifo': + os.mkfifo(path) + else: + path.mkdir() + try: + with self.assertRaisesRegex(native.ProbeError, 'lock_unavailable'): + native.scan(101, self.runner()) + self.assertEqual(self.calls, []) + self.assertEqual(target.read_bytes(), b'keep') + finally: + path.rmdir() if kind == 'directory' else path.unlink() + with patch.object(Path, 'open', side_effect=PermissionError): + with self.assertRaisesRegex(native.ProbeError, 'config_unavailable'): + native.scan(101, self.runner()) + self.assertEqual(self.calls, []) + def test_same_ct_busy_and_different_ct_cache_isolation(self): lock = threading.Lock() lock.acquire() @@ -546,7 +683,7 @@ class Routes(unittest.TestCase): '_VM_APP_SUGGESTIONS_TTL': 300, '_vm_cache_get': lambda cache, vmid, ttl: cache.get(vmid), '_vm_cache_put': lambda cache, vmid, value: cache.update({vmid: value}), - '_pvesh_cache': {}, '_vm_disk_cache': {}, '_vm_cache_invalidate': Mock(), + '_pvesh_cache': {}, '_vm_disk_cache': {}, '_vm_schedule_cache': {}, '_vm_cache_invalidate': Mock(), '_invalidate_lxc_ip': Mock(), '_schedule_started_guest_refresh': Mock(), '_publish_guest_modal_cache_revision': Mock()} tree = ast.parse((SCRIPTS / 'flask_server.py').read_text())