fix(monitor): place the Secure Gateway template on an active storage

AppImage/scripts/disk_temperature_history.py

@@ -410,60 +410,16 @@ def _extract_temperature(data: dict[str, Any]) -> Optional[float]:
 # ---------------------------------------------------------------------------
 
 
-# ── Leaving idle and excluded disks alone ──────────────────────────
+# ── Leaving excluded disks alone ─────────────────────────────────
 #
-# `-n standby` keeps a periodic reader from waking a disk that is asleep.
-# It does not let an awake one fall asleep: a drive's spin-down timer
-# counts time without commands, and a read every minute resets it, so an
-# unused drive whose timer is longer than that never spins down — and a
-# drive that parks its heads when idle loads them again on every read.
-#
-# So a rotational disk with no I/O since it was last looked at is not read
-# at all, not even asked for its power mode. The counters come from
-# /proc/diskstats, which costs no disk access, and a SMART query does not
-# move them — passthrough commands are not accounted as reads or writes —
-# so any change means something else used the disk. A disk in use is
-# already awake, and reading it then costs nothing. Solid-state disks have
-# no spindle and no heads, and keep their reading.
-#
-# A disk seen for the first time is read once, so a Monitor that has just
-# started still has values to show; after that it is left alone for as
-# long as nothing uses it.
+# A disk the user excludes in Settings receives nothing from a periodic
+# reader, not even the power-mode question, so it can spin down on its own
+# timer. Every other disk is read on schedule: `-n standby` still keeps a
+# reader from waking one that is already asleep.
 
 READ = "read"
-IDLE = "idle"
 EXCLUDED = "excluded"
 
-_last_io: dict[str, tuple[int, int]] = {}
-_idle_state: dict[str, float] = {}
-_IDLE_TTL = 600  # same horizon as the standby badge
-
-
-def _read_diskstats() -> dict[str, tuple[int, int]]:
-    """Reads and writes completed per device, from /proc/diskstats."""
-    out: dict[str, tuple[int, int]] = {}
-    try:
-        with open("/proc/diskstats") as f:
-            for line in f:
-                parts = line.split()
-                if len(parts) < 8:
-                    continue
-                try:
-                    out[parts[2]] = (int(parts[3]), int(parts[7]))
-                except ValueError:
-                    continue
-    except OSError:
-        pass
-    return out
-
-
-def _is_rotational(disk_name: str) -> bool:
-    try:
-        with open(f"/sys/block/{disk_name}/queue/rotational") as f:
-            return f.read().strip() == "1"
-    except OSError:
-        return False
-
 
 _EXCLUDED_TTL = 15
 _excluded_cache: Optional[tuple[float, set]] = None
@@ -503,50 +459,20 @@ _excluded_disk_names = excluded_disk_names
 
 
 def disk_read_policy(disk_name: str, excluded: Optional[set] = None) -> str:
-    """Whether a periodic reader may touch this disk now: READ, IDLE or
-    EXCLUDED. Shared by every reader that runs on its own, so the
-    temperature poller and the storage view cannot disagree about a disk.
-    ``excluded`` lets a caller that checks many disks pass the list once."""
+    """Whether a periodic reader may touch this disk: READ or EXCLUDED.
+    Shared by every reader that runs on its own, so the temperature poller
+    and the storage view cannot disagree about a disk. ``excluded`` lets a
+    caller that checks many disks pass the list once."""
     if excluded is None:
         excluded = _excluded_disk_names()
-    if disk_name in excluded:
-        _idle_state.pop(disk_name, None)
-        return EXCLUDED
-    if not _is_rotational(disk_name):
-        return READ
-    current = _read_diskstats().get(disk_name)
-    with _cache_lock:
-        previous = _last_io.get(disk_name)
-        if current is not None:
-            _last_io[disk_name] = current
-    if current is None or previous is None or current != previous:
-        _idle_state.pop(disk_name, None)
-        return READ
-    _idle_state[disk_name] = time.time()
-    return IDLE
-
-
-def settle_own_reads(disk_names) -> None:
-    """A reader that has just read these disks itself takes their counters as
-    the new baseline, so its own I/O never counts as activity."""
-    stats = _read_diskstats()
-    with _cache_lock:
-        for name in disk_names:
-            if name in _last_io and name in stats:
-                _last_io[name] = stats[name]
-
-
-def is_disk_idle(disk_name: str) -> bool:
-    """True while the disk is being left alone for having no I/O."""
-    ts = _idle_state.get(disk_name)
-    return ts is not None and (time.time() - ts) < _IDLE_TTL
+    return EXCLUDED if disk_name in excluded else READ
 
 
 def record_all_disk_temperatures() -> int:
     """Sample the disks that may be read now and persist their temperature.
 
-    USB disks are included. A disk the user excluded, or a rotational one
-    with no I/O since the last cycle, is skipped — see ``disk_read_policy``.
+    USB disks are included. A disk the user excluded is skipped — see
+    ``disk_read_policy``.
 
     Sampling fans out across a thread pool so a host with N disks pays
     roughly the time of the slowest single ``smartctl`` call instead of

AppImage/scripts/flask_health_routes.py

@@ -807,7 +807,6 @@ def get_disks_for_exclusion():
                 **disk,
                 'excluded': entry is not None,
                 'excluded_at': entry.get('excluded_at') if entry else None,
-                'idle': _dth.is_disk_idle(disk['name']),
                 'present': True,
             })
         # An excluded disk that is not connected right now — an unplugged USB
@@ -826,7 +825,6 @@ def get_disks_for_exclusion():
                 'rotational': False,
                 'excluded': True,
                 'excluded_at': entry.get('excluded_at'),
-                'idle': False,
                 'present': False,
             })
         result.sort(key=lambda d: (not d['present'], d['name']))

AppImage/scripts/flask_server.py

@@ -4418,9 +4418,6 @@ def get_system_disks():
             ['pvs', *dev_args, '--noheadings', '-o', 'pv_name,vg_name'],
             capture_output=True, text=True, timeout=5
         ) if dev_args is not None else None
-        if result is not None:
-            from lvm_device_scope import settle_scan_reads
-            settle_scan_reads()
         if result is not None and result.returncode == 0:
             for line in result.stdout.strip().split('\n'):
                 if line.strip():
@@ -4558,12 +4555,10 @@ def get_storage_info():
                         # temperature graph isn't a monitor bug — the
                         # disk is parked. See issue #232.
                         in_standby = False
-                        in_idle = False
                         in_excluded = False
                         try:
                             import disk_temperature_history as _dth
                             in_standby = _dth.is_disk_in_standby(disk_name)
-                            in_idle = _dth.is_disk_idle(disk_name)
                             in_excluded = disk_name in _dth.excluded_disk_names()
                         except Exception:
                             pass
@@ -4574,7 +4569,6 @@ def get_storage_info():
                             'size_bytes': disk_size_bytes,
                             'temperature': smart_data.get('temperature', 0),
                             'standby': in_standby,
-                            'idle': in_idle,
                             'excluded': in_excluded,
                             'health': smart_data.get('health', 'unknown'),
                             'power_on_hours': smart_data.get('power_on_hours', 0),
@@ -5305,28 +5299,19 @@ def get_smart_data(disk_name):
     if cached and now - cached[0] < _SMART_RESULT_TTL:
         return dict(cached[1])
 
-    # Excluded, or rotational with no I/O since it was last looked at: send
-    # it nothing — not even the power-mode question below, which is still a
-    # command. Serve what is known, without a temperature that would only be
-    # stale. Same rule as the temperature poller, so the two agree.
+    # Excluded in Settings: send it nothing — not even the power-mode
+    # question below, which is still a command. Serve what is known, without
+    # a temperature that would only be stale. Same rule as the temperature
+    # poller, so the two agree.
     try:
         import disk_temperature_history as _dth
         policy = _dth.disk_read_policy(disk_name)
     except Exception:
         policy = 'read'
     if policy != 'read':
-        if not cached and policy == 'idle' and not _hdd_in_standby(disk_name):
-            # Nothing is known of it yet, as after a restart: a disk that is
-            # spinning is read once; a parked one is not woken.
-            result = _get_smart_data_uncached(disk_name)
-            _dth.settle_own_reads([disk_name])
-            if _smart_data_useful(result):
-                _smart_result_cache[disk_name] = (now, dict(result))
-                cached = _smart_result_cache[disk_name]
         base = dict(cached[1]) if cached else _smart_identity_payload(disk_name)
         base['temperature'] = 0
-        base['excluded'] = policy == 'excluded'
-        base['idle'] = policy == 'idle'
+        base['excluded'] = True
         return base
 
     if _hdd_in_standby(disk_name):

AppImage/scripts/health_monitor.py

@@ -2288,9 +2288,6 @@ class HealthMonitor:
                 text=True,
                 timeout=3
             )
-            from lvm_device_scope import settle_scan_reads
-            settle_scan_reads()
-            
             if result.returncode != 0:
                 return {'status': 'WARNING', 'reason': 'lvs command failed'}
             
@@ -2314,7 +2311,6 @@ class HealthMonitor:
                     text=True,
                     timeout=3
                 )
-                settle_scan_reads()
                 if vg_result.returncode == 0 and vg_result.stdout.strip():
                     return {'status': 'WARNING', 'reason': 'No active LVM volumes detected'}
                 else:

AppImage/scripts/lvm_device_scope.py

@@ -9,20 +9,17 @@
 # Last Updated: 16/10/2026
 # ==========================================================
 # Description:
-# Applies the automatic disk-read policy to LVM metadata scans in both
-# Monitor readers. Excluded and idle disks receive no power-mode probe;
-# every device built on them is excluded too, and the scan's own reads
-# are not taken as activity. No eligible devices means no scan.
+# Applies the disk exclusions to LVM metadata scans in both Monitor
+# readers. Excluded disks receive no power-mode probe and a disk in
+# standby is not woken; every device built on them is excluded too. No
+# eligible devices means no scan.
 # ==========================================================
 
 import json
 import re
 import subprocess
-import threading
 
 _NAME = re.compile(r"[A-Za-z0-9._+-]+")
-_scan_lock = threading.Lock()
-_scanned_disks = set()
 
 
 def _topology():
@@ -116,14 +113,11 @@ def lvm_device_args(standby_check):
         excluded = excluded_disk_names()
         blocked = set()
         for disk in sorted(disks):
-            # Test policy first: even CHECK POWER MODE is forbidden for an
-            # excluded/idle disk. Some USB bridges do not honour -n standby.
+            # Test the exclusion first: even CHECK POWER MODE is forbidden
+            # for an excluded disk. Some USB bridges do not honour -n standby.
             if disk_read_policy(disk, excluded) != "read" or standby_check(disk):
                 blocked.add(disk)
         allowed = sorted(f"/dev/{name}" for name in volumes if not (_disks_under(name, parents) & blocked))
-        with _scan_lock:
-            _scanned_disks.clear()
-            _scanned_disks.update(disks - blocked)
         if not blocked:
             return []
         filters = _lvm_filters()
@@ -140,15 +134,3 @@ def lvm_device_args(standby_check):
         # Unavailable exclusions, topology or policy are not permission to
         # open devices that may have been deliberately excluded.
         return None
-
-
-def settle_scan_reads():
-    """The LVM scan reads the disks it is allowed to. Those reads are its
-    own, so they must not make an unused disk look busy at the next check."""
-    try:
-        from disk_temperature_history import settle_own_reads
-        with _scan_lock:
-            disks = set(_scanned_disks)
-        settle_own_reads(disks)
-    except Exception:
-        pass

AppImage/scripts/oci_manager.py

@@ -48,7 +48,8 @@ DEV_SCRIPTS_CATALOG = os.path.join(os.path.dirname(__file__), "..", "..", "oci",
 # Encryption key file
 ENCRYPTION_KEY_FILE = os.path.join(OCI_BASE_DIR, ".encryption_key")
 
-# Default storage for templates
+# Preferred storage for templates; another active one is used when it is
+# disabled, as `local` is on a BTRFS installation.
 DEFAULT_STORAGE = "local"
 
 # Default storage for rootfs (will be auto-detected)
@@ -389,7 +390,22 @@ def _host_arch() -> str:
     }.get(machine, 'amd64')
 
 
-def _download_alpine_template(storage: str = DEFAULT_STORAGE) -> bool:
+def _template_storage() -> str:
+    """An active storage that holds container templates: `local` when it
+    can, otherwise the first other one, such as `local-btrfs`."""
+    rc, out, _ = _run_pve_cmd(["pvesm", "status", "--content", "vztmpl"])
+    active = []
+    if rc == 0:
+        for line in out.strip().split("\n")[1:]:
+            parts = line.split()
+            if len(parts) >= 3 and parts[2] == "active":
+                active.append(parts[0])
+    if not active or DEFAULT_STORAGE in active:
+        return DEFAULT_STORAGE
+    return active[0]
+
+
+def _download_alpine_template(storage: Optional[str] = None) -> bool:
     """Download the latest Alpine LXC template using pveam.
 
     Filters by the host's architecture so an x86_64 host does not end up
@@ -397,6 +413,7 @@ def _download_alpine_template(storage: str = DEFAULT_STORAGE) -> bool:
     kept the last match and could pick any arch that `pveam available`
     happened to list (issue #324).
     """
+    storage = storage or _template_storage()
     print("[*] Downloading Alpine Linux template...")
     logger.info("Downloading Alpine template via pveam")
 
@@ -443,8 +460,9 @@ def _download_alpine_template(storage: str = DEFAULT_STORAGE) -> bool:
     return True
 
 
-def _find_alpine_template(storage: str = DEFAULT_STORAGE, auto_download: bool = True) -> Optional[str]:
+def _find_alpine_template(storage: Optional[str] = None, auto_download: bool = True) -> Optional[str]:
     """Find an available Alpine LXC template, optionally downloading if not found."""
+    storage = storage or _template_storage()
     template_dir = "/var/lib/vz/template/cache"
     
     # Try to get correct path from storage config
@@ -577,7 +595,7 @@ def _configure_tailscale(vmid: int, config: Dict[str, Any]) -> bool:
 # =================================================================
 # OCI Image Management
 # =================================================================
-def pull_oci_image(image: str, tag: str = "latest", storage: str = DEFAULT_STORAGE) -> Dict[str, Any]:
+def pull_oci_image(image: str, tag: str = "latest", storage: Optional[str] = None) -> Dict[str, Any]:
     """
     Pull an OCI image from a registry and store as LXC template.
     Uses Proxmox's pvesh API to download OCI images (same as GUI).
@@ -590,6 +608,7 @@ def pull_oci_image(image: str, tag: str = "latest", storage: str = DEFAULT_STORA
     Returns:
         Dict with success status and template path
     """
+    storage = storage or _template_storage()
     result = {
         "success": False,
         "message": "",

AppImage/scripts/tests/test_secure_gateway_template_storage.py

@@ -0,0 +1,35 @@
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+SCRIPTS_DIR = Path(__file__).resolve().parents[1]
+if str(SCRIPTS_DIR) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS_DIR))
+
+import oci_manager  # noqa: E402
+
+HEADER = "Name Type Status Total Used Available %\n"
+
+
+class TemplateStorage(unittest.TestCase):
+    """The Alpine template goes to a storage Proxmox can use."""
+
+    def storage(self, listing, rc=0):
+        with patch.object(oci_manager, "_run_pve_cmd", return_value=(rc, HEADER + listing, "")):
+            return oci_manager._template_storage()
+
+    def test_local_is_used_while_it_is_active(self):
+        self.assertEqual(self.storage("local dir active 1 1 1 1%\nnas nfs active 1 1 1 1%\n"), "local")
+
+    def test_another_storage_is_used_when_local_is_disabled(self):
+        # On a BTRFS installation `local` is disabled and not listed.
+        self.assertEqual(self.storage("local-btrfs btrfs active 1 1 1 1%\n"), "local-btrfs")
+        self.assertEqual(self.storage("nas nfs inactive 0 0 0 0%\nlocal-btrfs btrfs active 1 1 1 1%\n"), "local-btrfs")
+
+    def test_an_unreadable_status_keeps_local(self):
+        self.assertEqual(self.storage("", rc=1), "local")
+
+
+if __name__ == "__main__":
+    unittest.main()