leader discovery fix and code cleaning

src/storage/__init__.py

@@ -11,7 +11,7 @@ Also provides service discovery for inter-service communication.
 from .provider import StorageProvider, VideoMetadata
 from .redis_storage import RedisStorage
 from .leader_storage import LeaderStorage
-from .leader_discovery import LeaderDiscovery, LeaderLockInfo
+from .leader_client import LeaderClient, LeaderLockInfo
 from .service_discovery import (
     ServiceDiscovery,
     ServiceInfo,
@@ -24,7 +24,7 @@ __all__ = [
     'VideoMetadata',
     'RedisStorage',
     'LeaderStorage',
-    'LeaderDiscovery',
+    'LeaderClient',
     'LeaderLockInfo',
     'ServiceDiscovery',
     'ServiceInfo',

src/storage/leader_client.py

@@ -0,0 +1,260 @@
+"""
+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
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional, Callable
+from urllib.request import urlopen, Request
+from urllib.error import URLError
+
+# 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
+
+
+@dataclass
+class LeaderLockInfo:
+    """Leader lock file content format (kv-leader.info).
+    Written by meta-core leader, read by other services.
+    """
+    hostname: str
+    base_url: str
+    api_url: str      # meta-core API URL (port 9000)
+    redis_url: str
+    webdav_url: str
+    timestamp: int
+    pid: int
+
+
+@dataclass
+class URLsResponse:
+    """Response from meta-core /urls API."""
+    hostname: str
+    base_url: str
+    api_url: str
+    redis_url: str
+    webdav_url: 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.
+    """
+
+    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
+
+        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()
+
+    def get_leader_info(self) -> Optional[LeaderLockInfo]:
+        """Read leader info from the info file."""
+        try:
+            with open(self.info_file_path, 'r') as f:
+                content = f.read()
+
+            data = json.loads(content)
+
+            self.leader_info = LeaderLockInfo(
+                hostname=data.get('hostname', 'unknown'),
+                base_url=data.get('baseUrl', ''),
+                api_url=data.get('apiUrl', ''),
+                redis_url=data.get('redisUrl', ''),
+                webdav_url=data.get('webdavUrl', ''),
+                timestamp=data.get('timestamp', 0),
+                pid=data.get('pid', 0)
+            )
+
+            return self.leader_info
+
+        except FileNotFoundError:
+            print(f"[LeaderClient] Leader info file not found: {self.info_file_path}")
+            return None
+        except json.JSONDecodeError as e:
+            print(f"[LeaderClient] Error parsing leader info: {e}")
+            return None
+        except Exception as e:
+            print(f"[LeaderClient] Error reading 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."""
+        info = self.get_leader_info()
+        return info.webdav_url 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 leader info
+        if not api_url:
+            info = self.get_leader_info()
+            api_url = info.api_url if info else None
+
+        if not api_url:
+            print("[LeaderClient] No meta-core API URL available")
+            return None
+
+        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 get URLs: {response.status}")
+                    return None
+
+                data = json.loads(response.read().decode())
+
+                return 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', ''),
+                    is_leader=data.get('isLeader', False)
+                )
+
+        except URLError as e:
+            print(f"[LeaderClient] Error calling /urls API: {e}")
+            return None
+        except Exception as e:
+            print(f"[LeaderClient] Error getting URLs: {e}")
+            return None
+
+    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
+
+        while (time.time() - start_time) * 1000 < timeout_ms:
+            info = self.get_leader_info()
+            if info:
+                print(f"[LeaderClient] Leader found: {info.hostname} at {info.redis_url}")
+                return info
+
+            print("[LeaderClient] Waiting for leader...")
+            time.sleep(poll_interval)
+
+        raise TimeoutError(f"No leader found within {timeout_ms}ms")
+
+    def start_watching(self) -> None:
+        """Start watching for leader changes."""
+        if not WATCHDOG_AVAILABLE:
+            print("[LeaderClient] watchdog not available, file watching disabled")
+            return
+
+        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}")
+
+    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")
+
+    def _on_file_change(self) -> None:
+        """Handle file change event."""
+        print("[LeaderClient] Leader info changed, reloading...")
+        self.get_leader_info()
+        self._notify_change()
+
+    def on_change(self, callback: Callable[[], None]) -> 'LeaderClient':
+        """Register callback for leader changes."""
+        self._on_change_callbacks.append(callback)
+        return self
+
+    def _notify_change(self) -> None:
+        """Notify all change callbacks."""
+        for callback in self._on_change_callbacks:
+            try:
+                callback()
+            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
+
+    def close(self) -> None:
+        """Clean up resources."""
+        self._stop_event.set()
+        self.stop_watching()
+        self._on_change_callbacks = []

src/storage/leader_discovery.py

@@ -1,300 +0,0 @@
-"""
-Leader Discovery for meta-stremio (FOLLOWER-only)
-
-This is a Python implementation of the leader discovery pattern used in meta-fuse.
-meta-stremio never spawns Redis - it only connects to the leader's Redis instance.
-
-Architecture:
-- Reads info file at META_CORE_PATH/locks/kv-leader.info
-- The leader holds flock on kv-leader.lock, writes metadata to kv-leader.info
-- Parses leader info (Redis URL, HTTP URL)
-- Watches for leader changes via polling and file watching
-- Triggers reconnection callbacks on leader failure/change
-"""
-from __future__ import annotations
-
-import os
-import json
-import time
-import threading
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Optional, Callable
-from urllib.request import urlopen
-from urllib.error import URLError
-
-# 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
-
-
-@dataclass
-class LeaderLockInfo:
-    """Leader lock file content format."""
-    host: str
-    api: str  # Redis connection URL (e.g., redis://10.0.1.50:6379)
-    http: str  # HTTP API URL for the leader service
-    timestamp: int
-    pid: int
-
-
-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 LeaderDiscovery:
-    """
-    Leader Discovery for meta-stremio.
-
-    Discovers and monitors the KV leader via a lock file in the shared filesystem.
-    """
-
-    def __init__(
-        self,
-        meta_core_path: str,
-        health_check_interval: float = 5.0,
-        max_failures: int = 3
-    ):
-        self.meta_core_path = meta_core_path
-        self.health_check_interval = health_check_interval
-        self.max_failures = max_failures
-
-        self.lock_file_path = os.path.join(meta_core_path, 'locks', 'kv-leader.info')
-        self.leader_info: Optional[LeaderLockInfo] = None
-
-        self._is_started = False
-        self._stop_event = threading.Event()
-        self._health_check_thread: Optional[threading.Thread] = None
-        self._file_observer = None
-        self._consecutive_failures = 0
-
-        # Event callbacks
-        self._on_leader_found_callback: Optional[Callable[[LeaderLockInfo], None]] = None
-        self._on_leader_lost_callback: Optional[Callable[[], None]] = None
-
-    def start(self) -> None:
-        """Start watching for leader."""
-        if self._is_started:
-            print("[LeaderDiscovery] Already started")
-            return
-
-        print(f"[LeaderDiscovery] Starting, lock file: {self.lock_file_path}")
-
-        # Ensure lock directory exists
-        self._ensure_lock_dir()
-
-        # Try to read existing leader info
-        self._check_for_leader()
-
-        # Start file watcher (if watchdog is available)
-        self._start_file_watcher()
-
-        # Start health check thread
-        self._start_health_check()
-
-        self._is_started = True
-
-    def stop(self) -> None:
-        """Stop watching."""
-        if not self._is_started:
-            return
-
-        print("[LeaderDiscovery] Stopping...")
-
-        self._stop_event.set()
-
-        # Stop health check thread
-        if self._health_check_thread and self._health_check_thread.is_alive():
-            self._health_check_thread.join(timeout=2.0)
-
-        # Stop file observer
-        if self._file_observer:
-            self._file_observer.stop()
-            self._file_observer.join(timeout=2.0)
-            self._file_observer = None
-
-        self._is_started = False
-
-    def _ensure_lock_dir(self) -> None:
-        """Ensure lock directory exists."""
-        lock_dir = os.path.dirname(self.lock_file_path)
-        try:
-            os.makedirs(lock_dir, exist_ok=True)
-        except OSError as e:
-            print(f"[LeaderDiscovery] Could not create lock directory: {e}")
-
-    def _check_for_leader(self) -> None:
-        """Check for leader by reading lock file."""
-        try:
-            with open(self.lock_file_path, 'r') as f:
-                content = f.read()
-
-            data = json.loads(content)
-
-            # Validate leader info
-            if data and data.get('api') and data.get('http'):
-                info = LeaderLockInfo(
-                    host=data.get('host', 'unknown'),
-                    api=data['api'],
-                    http=data['http'],
-                    timestamp=data.get('timestamp', 0),
-                    pid=data.get('pid', 0)
-                )
-
-                is_new = (
-                    self.leader_info is None or
-                    self.leader_info.api != info.api
-                )
-
-                self.leader_info = info
-                self._consecutive_failures = 0
-
-                if is_new:
-                    print(f"[LeaderDiscovery] Leader found: {info.host} at {info.api}")
-                    if self._on_leader_found_callback:
-                        self._on_leader_found_callback(info)
-
-        except FileNotFoundError:
-            # Lock file doesn't exist - no leader yet
-            if self.leader_info:
-                print("[LeaderDiscovery] Lock file removed, leader lost")
-                self._handle_leader_lost()
-
-        except json.JSONDecodeError as e:
-            print(f"[LeaderDiscovery] Error parsing lock file: {e}")
-
-        except Exception as e:
-            print(f"[LeaderDiscovery] Error reading lock file: {e}")
-
-    def _is_leader_healthy(self) -> bool:
-        """Check if current leader is still healthy."""
-        if not self.leader_info:
-            return False
-
-        try:
-            health_url = f"{self.leader_info.http}/health"
-            with urlopen(health_url, timeout=5) as response:
-                return response.status == 200
-        except (URLError, TimeoutError, Exception):
-            return False
-
-    def _handle_leader_lost(self) -> None:
-        """Handle leader being lost."""
-        self.leader_info = None
-        if self._on_leader_lost_callback:
-            self._on_leader_lost_callback()
-
-    def _start_file_watcher(self) -> None:
-        """Start watching lock file for changes."""
-        if not WATCHDOG_AVAILABLE:
-            print("[LeaderDiscovery] watchdog not available, using polling only")
-            return
-
-        try:
-            lock_dir = os.path.dirname(self.lock_file_path)
-            handler = LockFileHandler(
-                callback=self._check_for_leader,
-                lock_filename='kv-leader.info'
-            )
-
-            self._file_observer = Observer()
-            self._file_observer.schedule(handler, lock_dir, recursive=False)
-            self._file_observer.start()
-            print(f"[LeaderDiscovery] File watcher started for {lock_dir}")
-
-        except Exception as e:
-            print(f"[LeaderDiscovery] Could not start file watcher: {e}")
-
-    def _start_health_check(self) -> None:
-        """Start health check loop."""
-        def health_check_loop():
-            while not self._stop_event.is_set():
-                try:
-                    if not self.leader_info:
-                        # No leader known, try to find one
-                        self._check_for_leader()
-                    else:
-                        # Check if current leader is healthy
-                        healthy = self._is_leader_healthy()
-
-                        if healthy:
-                            self._consecutive_failures = 0
-                        else:
-                            self._consecutive_failures += 1
-                            print(f"[LeaderDiscovery] Leader health check failed ({self._consecutive_failures}/{self.max_failures})")
-
-                            if self._consecutive_failures >= self.max_failures:
-                                print("[LeaderDiscovery] Leader appears dead, marking as lost")
-                                self._handle_leader_lost()
-
-                                # Try to find new leader
-                                self._check_for_leader()
-
-                except Exception as e:
-                    print(f"[LeaderDiscovery] Health check error: {e}")
-
-                self._stop_event.wait(self.health_check_interval)
-
-        self._health_check_thread = threading.Thread(
-            target=health_check_loop,
-            daemon=True,
-            name="leader-health-check"
-        )
-        self._health_check_thread.start()
-
-    # ========================================================================
-    # Event Registration
-    # ========================================================================
-
-    def on_leader_found(self, callback: Callable[[LeaderLockInfo], None]) -> 'LeaderDiscovery':
-        """Register callback for when leader is found/changed."""
-        self._on_leader_found_callback = callback
-
-        # If we already have leader info, call immediately
-        if self.leader_info:
-            callback(self.leader_info)
-
-        return self
-
-    def on_leader_lost(self, callback: Callable[[], None]) -> 'LeaderDiscovery':
-        """Register callback for when leader is lost."""
-        self._on_leader_lost_callback = callback
-        return self
-
-    # ========================================================================
-    # Getters
-    # ========================================================================
-
-    def get_leader_info(self) -> Optional[LeaderLockInfo]:
-        """Get current leader info."""
-        return self.leader_info
-
-    def get_redis_url(self) -> Optional[str]:
-        """Get Redis URL of current leader."""
-        return self.leader_info.api if self.leader_info else None
-
-    def has_leader(self) -> bool:
-        """Check if a leader is known."""
-        return self.leader_info is not None

src/storage/leader_storage.py

@@ -1,7 +1,7 @@
 """
 Leader-aware Storage Provider for meta-stremio.
 
-Uses leader discovery to automatically connect to the KV leader.
+Uses LeaderClient to read leader info from meta-core.
 Handles reconnection when the leader changes or becomes unavailable.
 """
 from __future__ import annotations
@@ -12,7 +12,7 @@ import threading
 from typing import List, Optional, Callable
 
 from .provider import StorageProvider, VideoMetadata
-from .leader_discovery import LeaderDiscovery, LeaderLockInfo
+from .leader_client import LeaderClient, LeaderLockInfo
 
 try:
     import redis
@@ -32,8 +32,8 @@ class LeaderStorage(StorageProvider):
     """
     Storage provider that discovers and connects to the KV leader.
 
-    Uses leader discovery to automatically find the Redis instance
+    Uses LeaderClient to read leader info from meta-core.
-    managed by meta-sort. Handles reconnection on leader failure.
+    Handles reconnection on leader failure.
     """
 
     def __init__(
@@ -51,7 +51,7 @@ class LeaderStorage(StorageProvider):
         self._prefix = prefix or REDIS_PREFIX
         self._override_url = redis_url  # Direct Redis URL (skip leader discovery)
 
-        self._leader_discovery: Optional[LeaderDiscovery] = None
+        self._leader_client: Optional[LeaderClient] = None
         self._client: Optional[redis.Redis] = None
         self._connected = False
         self._lock = threading.Lock()
@@ -61,31 +61,39 @@ class LeaderStorage(StorageProvider):
         self._on_disconnect_callbacks: List[Callable[[], None]] = []
 
     def connect(self) -> None:
-        """Connect to the storage backend via leader discovery."""
+        """Connect to the storage backend via leader client."""
         # If direct URL provided, skip leader discovery
         if self._override_url:
             print(f"[LeaderStorage] Using direct Redis URL: {self._override_url}")
             self._connect_to_redis(self._override_url)
             return
 
-        # Use leader discovery
+        # Use leader client
-        self._leader_discovery = LeaderDiscovery(
+        self._leader_client = LeaderClient(
             meta_core_path=self._meta_core_path
         )
 
-        # Set up callbacks
+        # Set up change callback
-        self._leader_discovery.on_leader_found(self._on_leader_found)
+        self._leader_client.on_change(self._on_leader_change)
-        self._leader_discovery.on_leader_lost(self._on_leader_lost)
 
-        # Start discovery
+        # Wait for leader and connect
-        self._leader_discovery.start()
+        try:
+            info = self._leader_client.wait_for_leader(30000)
+            print(f"[LeaderStorage] Connecting to leader at {info.redis_url}...")
+            self._connect_to_redis(info.redis_url)
+        except TimeoutError as e:
+            print(f"[LeaderStorage] Failed to find leader: {e}")
+            return
+
+        # Start watching for leader changes
+        self._leader_client.start_watching()
 
     def disconnect(self) -> None:
         """Disconnect from the storage backend."""
-        # Stop leader discovery
+        # Stop leader client
-        if self._leader_discovery:
+        if self._leader_client:
-            self._leader_discovery.stop()
+            self._leader_client.close()
-            self._leader_discovery = None
+            self._leader_client = None
 
         # Disconnect Redis
         self._disconnect_redis()
@@ -102,17 +110,21 @@ class LeaderStorage(StorageProvider):
                 self._connected = False
                 return False
 
-    def _on_leader_found(self, info: LeaderLockInfo) -> None:
+    def _on_leader_change(self) -> None:
-        """Handle leader found event."""
+        """Handle leader change event."""
-        print(f"[LeaderStorage] Connecting to leader at {info.api}...")
+        print("[LeaderStorage] Leader changed, reconnecting...")
-        self._connect_to_redis(info.api)
-
-    def _on_leader_lost(self) -> None:
-        """Handle leader lost event."""
-        print("[LeaderStorage] Leader lost, disconnecting...")
         self._disconnect_redis()
         self._notify_disconnect()
 
+        # Reconnect to new leader
+        if self._leader_client:
+            try:
+                info = self._leader_client.wait_for_leader(30000)
+                print(f"[LeaderStorage] Reconnecting to leader at {info.redis_url}...")
+                self._connect_to_redis(info.redis_url)
+            except TimeoutError as e:
+                print(f"[LeaderStorage] Failed to reconnect: {e}")
+
     def _connect_to_redis(self, url: str) -> None:
         """Connect to Redis."""
         with self._lock:
@@ -523,13 +535,15 @@ class LeaderStorage(StorageProvider):
     def get_status(self) -> dict:
         """Get storage status for dashboard."""
         leader_info = None
-        if self._leader_discovery:
+        if self._leader_client:
-            info = self._leader_discovery.get_leader_info()
+            info = self._leader_client.get_cached_leader_info()
             if info:
                 leader_info = {
-                    'host': info.host,
+                    'hostname': info.hostname,
-                    'api': info.api,
+                    'base_url': info.base_url,
-                    'http': info.http,
+                    'api_url': info.api_url,
+                    'redis_url': info.redis_url,
+                    'webdav_url': info.webdav_url,
                 }
 
         status = {

src/stremio.py

@@ -22,11 +22,10 @@ from urllib.parse import quote
 from typing import Optional, List, Dict
 from collections import defaultdict
 
-from storage import StorageProvider, VideoMetadata, RedisStorage, LeaderStorage
+from storage import StorageProvider, VideoMetadata, LeaderStorage
 
 # Configuration
 MEDIA_DIR = os.environ.get('MEDIA_DIR', '/files/watch')
-STORAGE_MODE = os.environ.get('STORAGE_MODE', 'redis').lower()  # 'leader' or 'redis'
 
 # Poster dimensions for Stremio
 POSTER_WIDTH = 342  # Stremio standard poster width
@@ -68,23 +67,15 @@ def get_poster_url(cid: Optional[str], width: Optional[int] = None) -> str:
 
 
 def init_storage() -> StorageProvider:
-    """Initialize the storage provider based on configuration.
+    """Initialize the storage provider.
 
-    Storage modes:
+    Uses LeaderStorage which reads leader info from meta-core.
-    - 'leader': Use leader discovery to find the KV store (recommended)
-    - 'redis': Direct Redis connection using REDIS_URL
     """
     global _storage
 
-    if STORAGE_MODE == 'redis':
+    _storage = LeaderStorage()
-        _storage = RedisStorage()
+    _storage.connect()
-        _storage.connect()
+    print("[Stremio] Using leader storage (reads leader info from meta-core)")
-        print(f"[Stremio] Using Redis storage")
-    else:
-        # Default to leader discovery
-        _storage = LeaderStorage()
-        _storage.connect()
-        print(f"[Stremio] Using leader discovery storage")
 
     return _storage