From 362be365528e17fc66eaf1be4c9b43fc18664e07 Mon Sep 17 00:00:00 2001 From: Worph Date: Sun, 13 Sep 2026 19:48:33 +0200 Subject: [PATCH] chnage service discovery --- src/server.py | 63 ++--- src/storage/leader_client.py | 410 ++++++++++++++----------------- src/storage/leader_storage.py | 11 +- src/storage/meshdisco.py | 437 ++++++++++++++++++++++++++++++++++ src/storage/meta_consumer.py | 12 +- www/meta-service-menu.js | 303 +++++++++++++++++++++++ 6 files changed, 973 insertions(+), 263 deletions(-) create mode 100644 src/storage/meshdisco.py create mode 100644 www/meta-service-menu.js diff --git a/src/server.py b/src/server.py index 9e61db8..a5f0e46 100644 --- a/src/server.py +++ b/src/server.py @@ -35,7 +35,6 @@ import threading import stremio import transcoder -from storage import init_service_discovery, get_service_discovery import fileserver import webdav_client @@ -65,13 +64,16 @@ DASHBOARD_PATHS = frozenset({ # Initialize storage storage = stremio.init_storage() -# Initialize service discovery (only if META_CORE_PATH exists) +# Service discovery is UDP now (meta-discovery v1), so there is no +# META_CORE_PATH gate: the leader client's node announces this service on the +# same multicast group it listens on. Being discoverable and discovering are +# one thing. See docs/project-architecture/service-discovery.md. service_discovery = None -if os.path.exists(META_CORE_PATH): - try: - service_discovery = init_service_discovery(base_url=BASE_URL if BASE_URL else None) - except Exception as e: - print(f"[Server] Service discovery init failed: {e}") +try: + from storage.leader_client import get_leader_client + get_leader_client().start_watching() +except Exception as e: + print(f"[Server] Service discovery init failed: {e}") # Global Stremio handler stremio_handler = stremio.StremioHandler() @@ -627,35 +629,35 @@ class Handler(BaseHTTPRequestHandler): self.wfile.write(content.encode('utf-8')) def handle_services_api(self): - """Return list of discovered services for dashboard navigation. + """Neighbours for the dashboard nav, from this peer's own UDP map. - Fetches services from meta-core (single source of truth) instead of - doing local filesystem-based discovery. + No hop through meta-core any more: every service keeps its own + neighbour table, so the nav still renders when meta-core is down. + `services` is kept as an alias of `neighbors` so the existing + dashboards keep working during the UI migration. """ from storage.leader_client import get_leader_client - from urllib.request import urlopen - from urllib.error import URLError try: - leader_client = get_leader_client() - leader_info = leader_client.get_leader_info() if leader_client else None - - if leader_info and leader_info.api_url: - # Fetch services from meta-core API - url = f"{leader_info.api_url}/services?current=meta-stremio" - with urlopen(url, timeout=5) as response: - data = json.loads(response.read().decode()) - return self.send_json(data) - except URLError as e: - print(f"[Server] Error fetching services from meta-core: {e}") + client = get_leader_client() + neighbors = client.neighbors() if client else [] + return self.send_json({ + 'current': 'meta-stremio', + 'enabled': True, + 'count': len(neighbors), + 'neighbors': neighbors, + 'services': neighbors, + 'self': client.self_announce() if client else None, + }) except Exception as e: print(f"[Server] Error in handle_services_api: {e}") - # Fallback: return empty list if meta-core is unavailable return self.send_json({ - 'services': [], 'current': 'meta-stremio', + 'enabled': False, 'count': 0, + 'neighbors': [], + 'services': [], }) def handle_file(self, cid: str, width: int = None): @@ -906,11 +908,14 @@ def shutdown_handler(signum, frame): """Handle graceful shutdown.""" print("\n[Server] Shutting down...") - # Stop service discovery + # Stop service discovery. close() emits a final "stopping" announce so + # neighbours drop us immediately instead of waiting out the staleness + # window. try: - sd = get_service_discovery() - if sd: - sd.stop() + from storage.leader_client import get_leader_client + client = get_leader_client() + if client: + client.close() except Exception as e: print(f"[Server] Error stopping service discovery: {e}") diff --git a/src/storage/leader_client.py b/src/storage/leader_client.py index 6b7429d..f57c404 100644 --- a/src/storage/leader_client.py +++ b/src/storage/leader_client.py @@ -1,40 +1,41 @@ +"""LeaderClient — locates meta-core and exposes its URLs. + +Since meta-discovery v1 this is a thin adapter over ``MetaCoreLocator`` +(``meshdisco.py``): meta-core is found by UDP announce instead of by reading +``/meta-core/locks/kv-leader.info``, so meta-stremio no longer needs the +``/meta-core`` volume mounted at all. + +The public surface is deliberately unchanged so ``LeaderStorage`` did not have +to move: ``get_leader_info``, ``get_urls``, ``get_api_url``, +``get_webdav_url*``, ``wait_for_leader``, ``start_watching`` / ``on_change`` +(the watchdog on the lock directory became "an announce arrived carrying a +different apiUrl", feeding the same callback). + +Precedence is the pin: when ``meta_core_url`` (``META_CORE_URL``) is set it +always wins and the wire is never consulted for core selection. See +docs/project-architecture/service-discovery.md. + +The name "leader" is legacy: meta-core's flock election is vestigial, and +``redis_url`` has been empty since the api-mediated-access lockdown — every +read goes over HTTP. """ -LeaderClient - Client for reading leader info from meta-core -This replaces the old LeaderDiscovery.py by delegating leader election -to meta-core (Go sidecar) and just reading the results. - -Features: -- Reads kv-leader.info file for leader URLs -- Calls meta-core /urls API for current leader URLs -- Watches for leader changes via file system watcher -""" -from __future__ import annotations - -import os import json -import threading +import os +import time +import urllib.request from dataclasses import dataclass -from pathlib import Path -from typing import Optional, Callable -from urllib.request import urlopen, Request -from urllib.error import URLError +from typing import Callable, List, Optional -# Try to import watchdog for file watching (optional) -try: - from watchdog.observers import Observer - from watchdog.events import FileSystemEventHandler - WATCHDOG_AVAILABLE = True -except ImportError: - WATCHDOG_AVAILABLE = False - Observer = None - FileSystemEventHandler = object +from .meshdisco import MetaCoreLocator @dataclass class LeaderLockInfo: - """Leader lock file content format (kv-leader.info). - Written by meta-core leader, read by other services. + """meta-core's identity and URLs. + + Historically the parsed kv-leader.info + /urls pair; now assembled from the + announce (or from /urls when meta-core is pinned). """ hostname: str base_url: str @@ -53,140 +54,133 @@ class URLsResponse: base_url: str api_url: str redis_url: str - webdav_url: str # External WebDAV URL (via nginx/HTTPS) - webdav_url_internal: str # Internal WebDAV URL (direct to port 9000) + webdav_url: str + webdav_url_internal: str is_leader: bool -class LockFileHandler(FileSystemEventHandler): - """File system event handler for lock file changes.""" - - def __init__(self, callback: Callable[[], None], lock_filename: str): - self.callback = callback - self.lock_filename = lock_filename - - def on_modified(self, event): - if event.src_path.endswith(self.lock_filename): - self.callback() - - def on_created(self, event): - if event.src_path.endswith(self.lock_filename): - self.callback() - - def on_deleted(self, event): - if event.src_path.endswith(self.lock_filename): - self.callback() - - class LeaderClient: - """ - LeaderClient for meta-stremio. - - Reads leader info from kv-leader.info file written by meta-core. - """ + """Locates meta-core over UDP and resolves its URL set.""" def __init__( self, - meta_core_path: str, - meta_core_url: Optional[str] = None - ): - self.meta_core_path = meta_core_path - self.meta_core_url = meta_core_url + meta_core_path: Optional[str] = None, + meta_core_url: Optional[str] = None, + service_name: str = "meta-stremio", + base_url: Optional[str] = None, + version: str = "", + ) -> None: + # meta_core_path is accepted and ignored — kept so existing callers + # construct unchanged. Nothing reads the volume any more. + self._meta_core_path = meta_core_path + self._meta_core_url = (meta_core_url or "").strip() or None + self._leader_info: Optional[LeaderLockInfo] = None + self._on_change_callbacks: List[Callable[[], None]] = [] + self._started = False - self.info_file_path = os.path.join(meta_core_path, 'locks', 'kv-leader.info') - self.leader_info: Optional[LeaderLockInfo] = None - - self._file_observer = None - self._on_change_callbacks: list[Callable[[], None]] = [] - self._stop_event = threading.Event() - - # URL caching + # Only used on the pinned path, where the URLs still come from an HTTP + # GET /urls. A discovered core carries them in the announce itself. self._cached_urls: Optional[URLsResponse] = None - self._urls_cache_time: float = 0 - self._urls_cache_ttl: float = 5.0 # 5 seconds + self._urls_cache_time: float = 0.0 + self._urls_cache_ttl: float = 5.0 - def _get_api_url_from_file(self) -> Optional[str]: - """Read API URL from file (plain text format).""" - try: - with open(self.info_file_path, 'r') as f: - content = f.read() - return content.strip() or None - except FileNotFoundError: - print(f"[LeaderClient] Leader info file not found: {self.info_file_path}") - return None - except Exception as e: - print(f"[LeaderClient] Error reading API URL from file: {e}") - return None + self._locator = MetaCoreLocator( + service_name=service_name, + base_url=base_url, + version=version, + meta_core_url=self._meta_core_url, + ) + self._locator.on_change(self._on_locator_change) + + # -- lifecycle --------------------------------------------------------- + + def _ensure_started(self) -> None: + if self._started: + return + self._started = True + self._locator.start() + + def _on_locator_change(self) -> None: + self._cached_urls = None + self._urls_cache_time = 0.0 + self._notify_change() + + # -- url resolution ---------------------------------------------------- + + def _assert_no_redis_url(self, redis_url: Optional[str]) -> None: + """Guard against a rollback reintroducing direct Redis exposure. + + api-mediated-access PR D removed redis_url; ALLOW_LEGACY_REDIS_URL=1 + downgrades this to a warning during a deliberate temporary rollback. + """ + if not redis_url: + return + msg = ( + "meta-core still publishes redisUrl; direct Redis access was retired " + "by the api-mediated-access lockdown. Verify meta-core version." + ) + if os.environ.get("ALLOW_LEGACY_REDIS_URL") == "1": + print(f"[LeaderClient] WARNING: {msg}") + else: + raise RuntimeError(msg) def _fetch_urls(self, api_url: str) -> Optional[URLsResponse]: - """Fetch URLs from meta-core /urls API with caching.""" - import time - - # Check cache now = time.time() if self._cached_urls and (now - self._urls_cache_time) < self._urls_cache_ttl: return self._cached_urls - try: - url = f"{api_url}/urls" - req = Request(url, headers={'Accept': 'application/json'}) - with urlopen(req, timeout=5) as response: - if response.status != 200: - print(f"[LeaderClient] Failed to fetch URLs: {response.status}") - return None - - data = json.loads(response.read().decode()) - - # api-mediated-access PR D: redisUrl is no longer published by - # meta-core. If it shows up, an old build snuck back in — - # fail loudly so the rollback can't go unnoticed. Set - # ALLOW_LEGACY_REDIS_URL=1 to downgrade to a warning during - # a deliberate temporary rollback. - if data.get('redisUrl'): - msg = ( - "[LeaderClient] meta-core still publishes redisUrl; " - "direct Redis access was retired by the api-mediated-access " - "lockdown. Verify meta-core version." - ) - if os.environ.get("ALLOW_LEGACY_REDIS_URL") == "1": - print("WARNING: " + msg) - else: - raise RuntimeError(msg) - self._cached_urls = URLsResponse( - hostname=data.get('hostname', ''), - base_url=data.get('baseUrl', ''), - api_url=data.get('apiUrl', ''), - redis_url=data.get('redisUrl', ''), - webdav_url=data.get('webdavUrl', ''), - webdav_url_internal=data.get('webdavUrlInternal', ''), - is_leader=data.get('isLeader', False) - ) - self._urls_cache_time = now - return self._cached_urls - - except URLError as e: + req = urllib.request.Request( + f"{api_url.rstrip('/')}/urls", headers={"Accept": "application/json"} + ) + with urllib.request.urlopen(req, timeout=5) as resp: + data = json.loads(resp.read().decode("utf-8")) + self._assert_no_redis_url(data.get("redisUrl")) + self._cached_urls = URLsResponse( + hostname=data.get("hostname", ""), + base_url=data.get("baseUrl", ""), + api_url=data.get("apiUrl", ""), + redis_url=data.get("redisUrl", "") or "", + webdav_url=data.get("webdavUrl", ""), + webdav_url_internal=data.get("webdavUrlInternal", ""), + is_leader=bool(data.get("isLeader", True)), + ) + self._urls_cache_time = now + return self._cached_urls + except Exception as e: print(f"[LeaderClient] Error calling /urls API: {e}") return None - except Exception as e: - print(f"[LeaderClient] Error fetching URLs: {e}") + + def get_urls(self) -> Optional[URLsResponse]: + """Resolve meta-core's URLs. + + Straight from the announce when discovered (no HTTP hop needed); over + HTTP when meta-core is pinned. + """ + self._ensure_started() + + wire = self._locator.get_urls() + if wire and not self._meta_core_url: + return URLsResponse( + hostname=wire.get("hostname", ""), + base_url=wire.get("baseUrl", ""), + api_url=wire.get("apiUrl", ""), + redis_url="", + webdav_url=wire.get("webdavUrl", ""), + webdav_url_internal=wire.get("webdavUrlInternal", ""), + is_leader=True, + ) + + api_url = self._locator.get_api_url() + if not api_url: return None + return self._fetch_urls(api_url) def get_leader_info(self) -> Optional[LeaderLockInfo]: - """Read leader info from file and /urls API.""" try: - # Read API URL from file (plain text) - api_url = self._get_api_url_from_file() - if not api_url: - return None - - # Fetch full info from /urls API - urls = self._fetch_urls(api_url) + urls = self.get_urls() if not urls: return None - - # Convert URLsResponse to LeaderLockInfo - import time - self.leader_info = LeaderLockInfo( + self._leader_info = LeaderLockInfo( hostname=urls.hostname, base_url=urls.base_url, api_url=urls.api_url, @@ -194,145 +188,107 @@ class LeaderClient: webdav_url=urls.webdav_url, webdav_url_internal=urls.webdav_url_internal, timestamp=int(time.time() * 1000), - pid=0 # Unknown for remote leader + pid=0, ) - - return self.leader_info - + return self._leader_info except Exception as e: - print(f"[LeaderClient] Error reading leader info: {e}") + print(f"[LeaderClient] Failed to read leader info: {e}") return None def get_redis_url(self) -> Optional[str]: - """Get Redis URL from leader info.""" info = self.get_leader_info() return info.redis_url if info else None def get_webdav_url(self) -> Optional[str]: - """Get WebDAV URL from leader info (external, via nginx/HTTPS).""" info = self.get_leader_info() return info.webdav_url if info else None def get_webdav_url_internal(self) -> Optional[str]: - """Get internal WebDAV URL from leader info (direct to port 9000). - Use this for container-to-container communication. - """ + """Internal WebDAV URL — use for container-to-container access.""" info = self.get_leader_info() return info.webdav_url_internal if info else None def get_api_url(self) -> Optional[str]: - """Get meta-core API URL from leader info.""" info = self.get_leader_info() return info.api_url if info else None - def get_urls(self) -> Optional[URLsResponse]: - """Call meta-core /urls API to get current URLs.""" - # First try using configured meta_core_url - api_url = self.meta_core_url - - # Fall back to reading from file - if not api_url: - api_url = self._get_api_url_from_file() - - if not api_url: - print("[LeaderClient] No meta-core API URL available") - return None - - return self._fetch_urls(api_url) - def wait_for_leader(self, timeout_ms: int = 30000) -> LeaderLockInfo: - """Wait for leader info to be available.""" - import time - start_time = time.time() - poll_interval = 0.5 # 500ms + """Block until meta-core is reachable. - while (time.time() - start_time) * 1000 < timeout_ms: + Probes immediately then every 500ms, matching the previous + file-polling loop's timing. + """ + self._ensure_started() + timeout_s = None if timeout_ms <= 0 else timeout_ms / 1000.0 + deadline = None if timeout_s is None else time.time() + timeout_s + + if self._locator.wait_for_core(timeout_s) is None: + raise TimeoutError(f"No meta-core found within {timeout_ms}ms") + + while deadline is None or time.time() < deadline: info = self.get_leader_info() if info: - print(f"[LeaderClient] Leader found: {info.hostname} at {info.redis_url}") + print(f"[LeaderClient] meta-core found: {info.hostname} at {info.api_url}") return info + time.sleep(0.5) + raise TimeoutError(f"No meta-core found within {timeout_ms}ms") - print("[LeaderClient] Waiting for leader...") - time.sleep(poll_interval) - - raise TimeoutError(f"No leader found within {timeout_ms}ms") + # -- change notification ---------------------------------------------- def start_watching(self) -> None: - """Start watching for leader changes.""" - if not WATCHDOG_AVAILABLE: - print("[LeaderClient] watchdog not available, file watching disabled") - return + """Previously a watchdog on the lock directory. - try: - lock_dir = os.path.dirname(self.info_file_path) - - handler = LockFileHandler( - callback=self._on_file_change, - lock_filename='kv-leader.info' - ) - - self._file_observer = Observer() - self._file_observer.schedule(handler, lock_dir, recursive=False) - self._file_observer.start() - print(f"[LeaderClient] Watching for leader changes in {lock_dir}") - - except Exception as e: - print(f"[LeaderClient] Could not start file watcher: {e}") + Discovery is always listening, so this only has to make sure the node + is running. + """ + self._ensure_started() def stop_watching(self) -> None: - """Stop watching for leader changes.""" - if self._file_observer: - self._file_observer.stop() - self._file_observer.join(timeout=2.0) - self._file_observer = None - print("[LeaderClient] Stopped watching for leader changes") + """No-op: the locator keeps listening until close().""" - def _on_file_change(self) -> None: - """Handle file change event.""" - print("[LeaderClient] Leader info changed, invalidating cache...") - # Invalidate cache to force fresh API call - self._cached_urls = None - self._urls_cache_time = 0 - - self.get_leader_info() - self._notify_change() - - def on_change(self, callback: Callable[[], None]) -> 'LeaderClient': - """Register callback for leader changes.""" + def on_change(self, callback: Callable[[], None]) -> "LeaderClient": self._on_change_callbacks.append(callback) return self def _notify_change(self) -> None: - """Notify all change callbacks.""" - for callback in self._on_change_callbacks: + print("[LeaderClient] meta-core changed, invalidating cache...") + for cb in list(self._on_change_callbacks): try: - callback() + cb() except Exception as e: print(f"[LeaderClient] Error in change callback: {e}") def get_cached_leader_info(self) -> Optional[LeaderLockInfo]: - """Get cached leader info (without re-reading file).""" - return self.leader_info + return self._leader_info + + # -- nav menu ---------------------------------------------------------- + + def neighbors(self) -> List[dict]: + """Neighbours for the nav menu (one row per service name).""" + self._ensure_started() + return self._locator.neighbors() + + def self_announce(self) -> dict: + """This service's own announce, so the menu can show itself.""" + self._ensure_started() + return self._locator.self_announce() def close(self) -> None: - """Clean up resources.""" - self._stop_event.set() - self.stop_watching() self._on_change_callbacks = [] + self._locator.stop() + self._started = False -# Singleton instance _leader_client: Optional[LeaderClient] = None def get_leader_client() -> Optional[LeaderClient]: - """Get or create the LeaderClient singleton. - - Returns None if META_CORE_PATH is not set. - """ + """Process-wide singleton, so one UDP socket serves every caller.""" global _leader_client if _leader_client is None: - meta_core_path = os.environ.get('META_CORE_PATH', '/meta-core') - meta_core_url = os.environ.get('META_CORE_URL') - _leader_client = LeaderClient(meta_core_path, meta_core_url) + _leader_client = LeaderClient( + meta_core_path=os.environ.get("META_CORE_PATH"), + meta_core_url=os.environ.get("META_CORE_URL"), + base_url=os.environ.get("BASE_URL") or None, + ) return _leader_client diff --git a/src/storage/leader_storage.py b/src/storage/leader_storage.py index 2cd8dce..525a43b 100644 --- a/src/storage/leader_storage.py +++ b/src/storage/leader_storage.py @@ -18,7 +18,7 @@ import time from typing import List, Optional, Callable from .provider import StorageProvider, VideoMetadata -from .leader_client import LeaderClient, LeaderLockInfo +from .leader_client import LeaderClient, LeaderLockInfo, get_leader_client from .meta_consumer import MetaConsumer from .meta_core_api_client import MetaCoreApiClient @@ -97,10 +97,11 @@ class LeaderStorage(StorageProvider): self._connect_to_redis(self._override_url) return - # Use leader client - self._leader_client = LeaderClient( - meta_core_path=self._meta_core_path - ) + # Use the process-wide leader client so meta-stremio holds ONE UDP + # discovery socket, shared with server.py's neighbours endpoint. + # META_CORE_URL, when set, pins meta-core and discovery never + # overrides it — see docs/project-architecture/service-discovery.md. + self._leader_client = get_leader_client() # Set up change callback self._leader_client.on_change(self._on_leader_change) diff --git a/src/storage/meshdisco.py b/src/storage/meshdisco.py new file mode 100644 index 0000000..ba114ee --- /dev/null +++ b/src/storage/meshdisco.py @@ -0,0 +1,437 @@ +"""meta-discovery v1 — UDP multicast service discovery. + +Replaces the two file-based mechanisms that required a shared /meta-core +volume: ``locks/kv-leader.info`` (locating meta-core at boot) and +``services/-.json`` (the dashboard nav registry). Both answered the +same question — where is meta-core, and who else is on this network — so one +announce packet answers both, and discovery scope becomes the docker network +rather than the mounted volume. + +The normative spec is docs/project-architecture/service-discovery.md. + +Two details this module must get right, both of which fail silently: + +1. **Per-interface fan-out.** A socket that joins with ``INADDR_ANY`` and sends + via the default route covers exactly one interface. Containers here are + routinely attached to two docker networks, so we enumerate interfaces and + join/send on each using a hand-packed ``ip_mreqn``. +2. **The pin wins.** When ``META_CORE_URL`` is set, the wire is never consulted + for core selection — otherwise a container that can see two cores could + latch onto the wrong one. +""" + +from __future__ import annotations + +import json +import os +import socket +import struct +import threading +import time +from typing import Callable, Dict, List, Optional + +PROTOCOL_VERSION = 1 +DEFAULT_GROUP = "239.255.77.1" +DEFAULT_PORT = 9399 +DEFAULT_INTERVAL = 10.0 +# A neighbour unseen for interval * this is dropped. Replaces the reaper. +LIVENESS_FACTOR = 3 +READ_BUFFER = 8192 + +TYPE_DISCOVERY = "discovery" +TYPE_ANNOUNCE = "announce" +ROLE_CORE = "core" +ROLE_SERVICE = "service" + + +def _env_flag(name: str, default: bool = True) -> bool: + raw = os.environ.get(name) + if raw is None or raw == "": + return default + return raw.strip().lower() not in ("false", "0", "no", "off") + + +def _interface_indexes() -> List[tuple]: + """(name, index, ipv4) for every up, non-loopback interface. + + Falls back to a single unspecified entry when the platform cannot + enumerate, so discovery still works on the default route. + """ + out: List[tuple] = [] + try: + for idx, name in socket.if_nameindex(): + if name == "lo": + continue + try: + # Cheap way to get the interface's IPv4 without netifaces. + addr = _ipv4_of(name) + except OSError: + continue + if addr: + out.append((name, idx, addr)) + except (AttributeError, OSError): + pass + return out + + +def _ipv4_of(ifname: str) -> Optional[str]: + """SIOCGIFADDR — avoids a netifaces dependency.""" + import fcntl + + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + packed = struct.pack("256s", ifname[:15].encode("utf-8")) + return socket.inet_ntoa(fcntl.ioctl(s.fileno(), 0x8915, packed)[20:24]) + except OSError: + return None + finally: + s.close() + + +def local_ipv4() -> str: + """First non-loopback IPv4, else the hostname. Matches the Go helper.""" + ifaces = _interface_indexes() + if ifaces: + return ifaces[0][2] + return socket.gethostname() + + +class MeshNode: + """Announces this service, answers probes, tracks neighbours.""" + + def __init__( + self, + name: str, + role: str = ROLE_SERVICE, + instance: Optional[str] = None, + version: str = "", + group: str = DEFAULT_GROUP, + port: int = DEFAULT_PORT, + interval: float = DEFAULT_INTERVAL, + enabled: bool = True, + payload: Optional[Callable[[], dict]] = None, + ) -> None: + self.name = name + self.role = role + self.instance = instance or socket.gethostname() + self.version = version + self.group = group + self.port = port + self.interval = interval + self.enabled = enabled + self._payload = payload + + self._sock: Optional[socket.socket] = None + self._ifaces: List[tuple] = [] + self._neighbors: Dict[str, dict] = {} + self._lock = threading.Lock() + self._send_lock = threading.Lock() + self._running = False + self._threads: List[threading.Thread] = [] + self._listeners: List[Callable[[dict], None]] = [] + + # -- lifecycle --------------------------------------------------------- + + def start(self) -> None: + if self._running or not self.enabled: + if not self.enabled: + print("[meshdisco] Disabled by configuration") + return + + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) + except (AttributeError, OSError): + pass + sock.bind(("0.0.0.0", self.port)) + # TTL 1: link-local only, never routed. + sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 1) + + self._ifaces = _interface_indexes() + group_bin = socket.inet_aton(self.group) + joined = 0 + for ifname, idx, addr in self._ifaces: + try: + # ip_mreqn: multiaddr, address (INADDR_ANY), ifindex. The + # ifindex is what makes this a per-interface join instead of + # "whatever the routing table picks". + mreq = group_bin + socket.inet_aton("0.0.0.0") + struct.pack("@i", idx) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) + joined += 1 + except OSError as e: + print(f"[meshdisco] Warning: join {self.group} on {ifname} failed: {e}") + + self._sock = sock + self._running = True + print( + f"[meshdisco] Listening on {self.group}:{self.port} as " + f"{self.name}/{self.instance} (joined {joined} of {len(self._ifaces)} interfaces)" + ) + + for target in (self._read_loop, self._announce_loop): + t = threading.Thread(target=target, daemon=True) + t.start() + self._threads.append(t) + + def stop(self) -> None: + if not self._running: + return + self._running = False + # Final announce so neighbours drop us immediately rather than waiting + # out the staleness window. + try: + self._send(self._build_announce("stopping")) + except OSError: + pass + if self._sock: + try: + self._sock.close() + except OSError: + pass + self._sock = None + + # -- api --------------------------------------------------------------- + + def probe(self) -> None: + """Multicast a probe; every listener replies immediately.""" + self._send({"v": PROTOCOL_VERSION, "type": TYPE_DISCOVERY}) + + def neighbors(self) -> List[dict]: + cutoff = time.time() - self.interval * LIVENESS_FACTOR + with self._lock: + for key in [k for k, v in self._neighbors.items() if v["lastSeen"] < cutoff]: + del self._neighbors[key] + out = list(self._neighbors.values()) + return sorted(out, key=lambda n: n.get("name") or "") + + def neighbors_by_name(self) -> List[dict]: + best: Dict[str, dict] = {} + for nb in self.neighbors(): + cur = best.get(nb.get("name") or "") + if cur is None or nb["lastSeen"] > cur["lastSeen"]: + best[nb.get("name") or ""] = nb + return sorted(best.values(), key=lambda n: n.get("name") or "") + + def cores(self) -> List[dict]: + return [n for n in self.neighbors() if n.get("role") == ROLE_CORE and n.get("urls")] + + def self_announce(self) -> dict: + msg = self._build_announce("running") + msg["addr"] = "" + msg["lastSeen"] = time.time() + return msg + + def on_announce(self, cb: Callable[[dict], None]) -> None: + self._listeners.append(cb) + + # -- internals --------------------------------------------------------- + + def _build_announce(self, status: str) -> dict: + msg = { + "v": PROTOCOL_VERSION, + "type": TYPE_ANNOUNCE, + "name": self.name, + "instance": self.instance, + "role": self.role, + "version": self.version, + "status": status, + } + if self._payload: + try: + p = self._payload() or {} + except Exception: + p = {} + if p.get("baseUrl"): + msg["baseUrl"] = p["baseUrl"] + if p.get("status") and status == "running": + msg["status"] = p["status"] + # Only a core may carry a URLs block. + if self.role == ROLE_CORE and p.get("urls"): + msg["urls"] = p["urls"] + return msg + + def _send(self, msg: dict, dest: Optional[tuple] = None) -> None: + sock = self._sock + if sock is None: + return + body = json.dumps(msg).encode("utf-8") + with self._send_lock: + if dest is not None: + try: + sock.sendto(body, dest) + except OSError as e: + print(f"[meshdisco] Reply to {dest} failed: {e}") + return + # One write per interface — the default route alone reaches a + # single network. + for ifname, idx, addr in self._ifaces: + try: + sock.setsockopt( + socket.IPPROTO_IP, + socket.IP_MULTICAST_IF, + socket.inet_aton(addr) + socket.inet_aton("0.0.0.0") + struct.pack("@i", idx), + ) + sock.sendto(body, (self.group, self.port)) + except OSError as e: + print(f"[meshdisco] Warning: send on {ifname} failed: {e}") + + def _read_loop(self) -> None: + while self._running: + sock = self._sock + if sock is None: + return + try: + data, addr = sock.recvfrom(READ_BUFFER) + except OSError: + if not self._running: + return + continue + try: + msg = json.loads(data.decode("utf-8")) + except (ValueError, UnicodeDecodeError): + continue # foreign traffic on the group + if msg.get("v") != PROTOCOL_VERSION: + continue + # Drop our own multicast echo. + if msg.get("instance") == self.instance and msg.get("name") == self.name: + continue + + if msg.get("type") == TYPE_DISCOVERY: + self._send(self._build_announce("running"), dest=addr) + continue + if msg.get("type") != TYPE_ANNOUNCE or not msg.get("name"): + continue + + key = f"{msg['name']}|{msg.get('instance', '')}" + if msg.get("status") == "stopping": + with self._lock: + self._neighbors.pop(key, None) + continue + + # Source address comes from the packet, never the payload. + msg["addr"] = addr[0] + msg["lastSeen"] = time.time() + with self._lock: + self._neighbors[key] = msg + for cb in list(self._listeners): + try: + cb(msg) + except Exception as e: + print(f"[meshdisco] Listener raised: {e}") + + def _announce_loop(self) -> None: + self._send(self._build_announce("running")) + self.probe() + while self._running: + time.sleep(self.interval) + if not self._running: + return + try: + self._send(self._build_announce("running")) + except OSError: + pass + + +class MetaCoreLocator: + """Locates meta-core over UDP, replacing the kv-leader.info read.""" + + def __init__( + self, + service_name: str, + base_url: Optional[str] = None, + version: str = "", + meta_core_url: Optional[str] = None, + enabled: Optional[bool] = None, + ) -> None: + self._pin = (meta_core_url or "").strip() or None + self._current: Optional[dict] = None + self._change_callbacks: List[Callable[[], None]] = [] + + self._node = MeshNode( + name=service_name, + role=ROLE_SERVICE, + version=version, + enabled=_env_flag("ENABLE_UDP_DISCOVERY", True) if enabled is None else enabled, + payload=lambda: {"baseUrl": base_url or f"http://{local_ipv4()}"}, + ) + self._node.on_announce(self._on_announce) + + def _on_announce(self, nb: dict) -> None: + if nb.get("role") != ROLE_CORE or not nb.get("urls"): + return + if self._pin: + return # pinned: the wire cannot move us + + prev = self._current + if prev is None: + self._current = nb["urls"] + print(f"[meshdisco] meta-core discovered at {nb['urls']['apiUrl']} ({nb.get('addr')})") + for cb in list(self._change_callbacks): + try: + cb() + except Exception as e: + print(f"[meshdisco] Change callback raised: {e}") + return + if prev.get("apiUrl") == nb["urls"].get("apiUrl"): + self._current = nb["urls"] + return + # A second, different core. Do not flap — keep the first and say so, + # because on a PCS box this means a client container can see a core it + # must not use. + print( + f"[meshdisco] Ignoring a second meta-core at {nb['urls']['apiUrl']} " + f"({nb.get('addr')}); staying with {prev.get('apiUrl')}. " + f"Set META_CORE_URL to pin this explicitly." + ) + + def start(self) -> None: + if self._pin: + print(f"[meshdisco] meta-core pinned to {self._pin}; discovery is advisory") + self._node.start() + + def stop(self) -> None: + self._node.stop() + self._change_callbacks = [] + + def get_api_url(self) -> Optional[str]: + if self._pin: + return self._pin + if self._current: + return self._current.get("apiUrl") + return None + + def get_urls(self) -> Optional[dict]: + return self._current + + def wait_for_core(self, timeout: Optional[float] = 30.0) -> Optional[str]: + """Block until meta-core is located. + + ``timeout=None`` waits forever, which is meta-stremio's default + (LEADER_WAIT_TIMEOUT=0). Re-probes every 500ms rather than waiting out + an announce interval. + """ + deadline = None if timeout is None else time.time() + timeout + logged = False + while deadline is None or time.time() < deadline: + url = self.get_api_url() + if url: + return url + if not logged: + print("[meshdisco] Waiting for meta-core...") + logged = True + self._node.probe() + time.sleep(0.5) + return None + + def on_change(self, cb: Callable[[], None]) -> None: + self._change_callbacks.append(cb) + + def neighbors(self) -> List[dict]: + return self._node.neighbors_by_name() + + def self_announce(self) -> dict: + return self._node.self_announce() + + def probe(self) -> None: + self._node.probe() diff --git a/src/storage/meta_consumer.py b/src/storage/meta_consumer.py index aaa5eae..e87242e 100644 --- a/src/storage/meta_consumer.py +++ b/src/storage/meta_consumer.py @@ -50,11 +50,19 @@ class MetaConsumer: def __init__( self, api_url: str, - cursor_path: str = "/meta-core/cursors/meta-stremio-meta.cursor", + cursor_path: Optional[str] = None, request_timeout: int = 35, ) -> None: self._api_url = api_url.rstrip("/") - self._cursor_path = cursor_path + # The SSE resume cursor used to default under /meta-core, which is not + # mounted any more (meta-discovery v1 removed the shared volume). It + # belongs with the rest of this service's local state: CACHE_FOLDER_PATH + # (/data/cache in the dev stack). Without a writable home the consumer + # silently replays from `$` on every restart. + self._cursor_path = cursor_path or os.path.join( + os.environ.get("CACHE_FOLDER_PATH", "/data/cache"), + "meta-stremio-meta.cursor", + ) self._request_timeout = request_timeout self._running = False self._thread: Optional[threading.Thread] = None diff --git a/www/meta-service-menu.js b/www/meta-service-menu.js new file mode 100644 index 0000000..17fa6b2 --- /dev/null +++ b/www/meta-service-menu.js @@ -0,0 +1,303 @@ +/** + * — the neighbouring meta-* services, as a burger menu. + * + * Replaces the eight divergent "services bar" implementations (four copies of + * a React `ServiceNav.tsx`, two mirrored `services.js`, two bespoke inline + * ones) with a single framework-free custom element. It drops into the React + * dashboards as a plain tag — no React dependency — and into the vanilla UIs + * as an ES module. + * + * Data comes from the host service's own `GET /api/neighbors`, same-origin. + * Never cross-origin: the dev stack's self-signed certs and the OIDC redirects + * make a fetch to a neighbour a dead end. Links are plain navigations. + * + * ⚠ MIRRORED FILE. Byte-identical copies live in every meta-* UI tree; see + * `scripts/check-mirrors.sh`. Edit one → edit all. + * + * Shadow DOM is not optional here. The eight UIs use four incompatible token + * vocabularies, and two of them define *different colours under the same + * names* (`--bg-secondary` is #12121a in meta-sort, #16213e in meta-core, + * #161b22 in meta-dup). meta-search and meta-share are also the only + * light-mode-capable UIs, so a hardcoded dark palette would break them. + * The element therefore styles itself from four tokens with dark defaults: + * + * --mm-nav-fg, --mm-nav-bg, --mm-nav-border, --mm-nav-accent + * + * Each host maps its own variables onto those, e.g. + * meta-service-menu { --mm-nav-bg: var(--bg-secondary); } + * + * Attributes: + * current name of the host service, so it can be marked and not linked + * endpoint override the default "/api/neighbors" + * label button tooltip (default "Services") + * + * Property: + * rewriteUrl optional (name, url, neighbour) => url, for hosts that must + * translate internal hostnames into browser-reachable ones + * (meta-gateway does this for its dev/prod split). + */ + +const ICONS = { + "meta-core": "⚙️", + "meta-sort": "📁", + "meta-fuse": "🗂️", + "meta-stremio": "🎬", + "meta-dup": "🔍", + "meta-search": "🔎", + "meta-share": "🔗", + "meta-gateway": "🌉", + "meta-watch": "📺", + "meta-listen": "🎧", + "meta-read": "📖", +}; +const DEFAULT_ICON = "📦"; +const POLL_MS = 30000; + +function titleCase(name) { + return String(name || "") + .split("-") + .map((w) => w.charAt(0).toUpperCase() + w.slice(1)) + .join(" "); +} + +const TEMPLATE = ` + + + +`; + +class MetaServiceMenu extends HTMLElement { + constructor() { + super(); + this.attachShadow({ mode: "open" }); + this.shadowRoot.innerHTML = TEMPLATE; + this._btn = this.shadowRoot.querySelector("button.trigger"); + this._panel = this.shadowRoot.querySelector(".panel"); + this._neighbors = []; + this._timer = null; + this._open = false; + + this._onDocClick = (e) => { + if (!this._open) return; + if (e.composedPath().includes(this)) return; + this._setOpen(false); + }; + this._onKey = (e) => { + if (e.key === "Escape" && this._open) { + this._setOpen(false); + this._btn.focus(); + } + }; + } + + connectedCallback() { + this._btn.title = this.getAttribute("label") || "Services"; + this._btn.setAttribute("aria-label", this._btn.title); + this._btn.addEventListener("click", () => this._setOpen(!this._open)); + document.addEventListener("click", this._onDocClick); + document.addEventListener("keydown", this._onKey); + + this.refresh(); + this._timer = setInterval(() => this.refresh(), POLL_MS); + } + + disconnectedCallback() { + if (this._timer) clearInterval(this._timer); + this._timer = null; + document.removeEventListener("click", this._onDocClick); + document.removeEventListener("keydown", this._onKey); + } + + get endpoint() { + return this.getAttribute("endpoint") || "/api/neighbors"; + } + + get current() { + return this.getAttribute("current") || ""; + } + + async refresh() { + try { + const res = await fetch(this.endpoint, { + headers: { Accept: "application/json" }, + }); + if (!res.ok) throw new Error(String(res.status)); + const data = await res.json(); + // `services` is the legacy alias some backends still emit. + this._neighbors = data.neighbors || data.services || []; + } catch { + // Silent: the menu is navigation, not a health indicator. Keep whatever + // we last had rather than blanking the list on one failed poll. + } + this._render(); + } + + _setOpen(open) { + this._open = open; + this._panel.hidden = !open; + this._btn.setAttribute("aria-expanded", String(open)); + if (open) this.refresh(); + } + + _render() { + const panel = this._panel; + panel.textContent = ""; + + const seen = new Set(); + const rows = []; + for (const n of this._neighbors) { + const name = n.name || ""; + if (!name || seen.has(name)) continue; + seen.add(name); + rows.push(n); + } + // Always show the host itself, even before its own echo comes back. + if (this.current && !seen.has(this.current)) { + rows.push({ name: this.current, baseUrl: "" }); + } + rows.sort((a, b) => String(a.name).localeCompare(String(b.name))); + + if (rows.length === 0) { + const empty = document.createElement("div"); + empty.className = "empty"; + empty.textContent = "No services found"; + panel.appendChild(empty); + return; + } + + for (const n of rows) { + const isActive = n.name === this.current; + let url = n.baseUrl || ""; + if (typeof this.rewriteUrl === "function") { + try { + url = this.rewriteUrl(n.name, url, n) || ""; + } catch { + /* a bad hook must not blank the menu */ + } + } + + // A neighbour with no reachable URL is shown but not linked — better + // than a link that goes nowhere. + const el = document.createElement(isActive || !url ? "span" : "a"); + el.className = + "item" + (isActive ? " active" : "") + (!isActive && !url ? " dead" : ""); + if (el.tagName === "A") { + el.href = url; + el.setAttribute("role", "menuitem"); + } + if (!isActive && !url) el.title = "No reachable URL announced"; + + const icon = document.createElement("span"); + icon.className = "icon"; + icon.textContent = ICONS[n.name] || DEFAULT_ICON; + + const label = document.createElement("span"); + label.textContent = titleCase(n.name); + + el.append(icon, label); + if (n.status === "running" || isActive) { + const dot = document.createElement("span"); + dot.className = "dot"; + dot.title = n.status || "running"; + el.appendChild(dot); + } + panel.appendChild(el); + } + } +} + +if (!customElements.get("meta-service-menu")) { + customElements.define("meta-service-menu", MetaServiceMenu); +} + +export default MetaServiceMenu;