feat: expose per-group refresh plans and cache storage daily

app/storage_cache.py

@@ -0,0 +1,57 @@
+"""Daily filesystem capacity cache, persisted in RAM across monitor restarts."""
+import json
+import time
+from pathlib import Path
+from shutil import disk_usage
+
+
+class StorageCache:
+    def __init__(self, path, reader=disk_usage, clock=time.time):
+        self.path, self.reader, self.clock = Path(path), reader, clock
+        self.values = None
+        self.observed_at = None
+        self.next_update_at = 0
+        self.error = False
+        try:
+            data = json.loads(self.path.read_text())
+            if not isinstance(data,dict):raise ValueError('invalid cache')
+            retry = data.get('next_update_at')
+            if data.get('error') is True and isinstance(retry,(int,float)) and self.clock() < retry <= self.clock()+3600:
+                self.error, self.next_update_at = True, retry
+            values = data['values']
+            if all(isinstance(values[k], int) and not isinstance(values[k], bool) and values[k] >= 0 for k in ('total', 'used', 'free')) and values['total'] > 0 and values['used'] <= values['total']:
+                at = data['observed_at']
+                if isinstance(at, (int, float)) and 0 <= self.clock() - at < 90000:
+                    self.values, self.observed_at = values, at
+                    if not self.error:self.next_update_at = at + 86400
+        except (OSError, ValueError, KeyError, TypeError):
+            pass
+
+    def sample(self):
+        now = self.clock()
+        if now < self.next_update_at:
+            return self.values
+        try:
+            usage = self.reader('/')
+            if usage.total <= 0 or not 0 <= usage.used <= usage.total or usage.free < 0:
+                raise ValueError('invalid capacity')
+        except (OSError, ValueError):
+            self.error = True
+            self.next_update_at = now + 3600
+            self.persist()
+            return self.values
+        self.values = dict(total=usage.total, used=usage.used, free=usage.free)
+        self.observed_at = now
+        self.next_update_at = now + 86400
+        self.error = False
+        self.persist()
+        return self.values
+
+    def persist(self):
+        try:
+            self.path.parent.mkdir(parents=True, exist_ok=True)
+            self.path.write_text(json.dumps(dict(values=self.values, observed_at=self.observed_at,
+                                                next_update_at=self.next_update_at, error=self.error)))
+        except OSError:
+            # The in-memory sample remains valid even if the RAM cache cannot be persisted.
+            pass