From f26fea797c30b5e16d1a0f3d26e86891fecbe4cb Mon Sep 17 00:00:00 2001 From: Pierre Date: Sat, 21 Feb 2026 18:10:04 +0100 Subject: [PATCH] fix leader discovery --- src/storage/leader_client.py | 17 +++- src/storage/leader_storage.py | 167 +++++++++++++++++++++++++++++----- 2 files changed, 157 insertions(+), 27 deletions(-) diff --git a/src/storage/leader_client.py b/src/storage/leader_client.py index b391130..92dfa0c 100644 --- a/src/storage/leader_client.py +++ b/src/storage/leader_client.py @@ -40,7 +40,8 @@ class LeaderLockInfo: base_url: str api_url: str # meta-core API URL (port 9000) redis_url: str - webdav_url: str + webdav_url: str # External WebDAV URL (via nginx/HTTPS) + webdav_url_internal: str # Internal WebDAV URL (direct to port 9000) timestamp: int pid: int @@ -52,7 +53,8 @@ class URLsResponse: base_url: str api_url: str redis_url: str - webdav_url: str + webdav_url: str # External WebDAV URL (via nginx/HTTPS) + webdav_url_internal: str # Internal WebDAV URL (direct to port 9000) is_leader: bool @@ -141,6 +143,7 @@ class LeaderClient: 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 @@ -174,6 +177,7 @@ class LeaderClient: api_url=urls.api_url, redis_url=urls.redis_url, webdav_url=urls.webdav_url, + webdav_url_internal=urls.webdav_url_internal, timestamp=int(time.time() * 1000), pid=0 # Unknown for remote leader ) @@ -190,10 +194,17 @@ class LeaderClient: return info.redis_url if info else None def get_webdav_url(self) -> Optional[str]: - """Get WebDAV URL from leader info.""" + """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. + """ + 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() diff --git a/src/storage/leader_storage.py b/src/storage/leader_storage.py index 8078dcc..4cee553 100644 --- a/src/storage/leader_storage.py +++ b/src/storage/leader_storage.py @@ -3,12 +3,18 @@ Leader-aware Storage Provider for meta-stremio. Uses LeaderClient to read leader info from meta-core. Handles reconnection when the leader changes or becomes unavailable. + +Features: +- Waits for meta-core leader before accepting requests +- Automatic reconnection on leader change or failure +- Configurable startup timeout via LEADER_WAIT_TIMEOUT env var """ from __future__ import annotations import os import json import threading +import time from typing import List, Optional, Callable from .provider import StorageProvider, VideoMetadata @@ -31,6 +37,13 @@ META_CORE_PATH = os.environ.get('META_CORE_PATH', '/meta-core') FILES_PATH = os.environ.get('FILES_PATH', '/files') REDIS_PREFIX = os.environ.get('REDIS_PREFIX', '') +# Leader wait timeout: how long to wait for leader on startup (in seconds) +# 0 = wait forever (default), >0 = timeout after N seconds +LEADER_WAIT_TIMEOUT = int(os.environ.get('LEADER_WAIT_TIMEOUT', '0')) + +# Retry interval for leader connection attempts (in seconds) +LEADER_RETRY_INTERVAL = int(os.environ.get('LEADER_RETRY_INTERVAL', '5')) + class LeaderStorage(StorageProvider): """ @@ -70,7 +83,12 @@ class LeaderStorage(StorageProvider): self._on_disconnect_callbacks: List[Callable[[], None]] = [] def connect(self) -> None: - """Connect to the storage backend via leader client.""" + """Connect to the storage backend via leader client. + + This method blocks until a leader is found and connection is established. + If LEADER_WAIT_TIMEOUT > 0, it will timeout after that many seconds. + If LEADER_WAIT_TIMEOUT == 0 (default), it will wait forever. + """ # If direct URL provided, skip leader discovery if self._override_url: print(f"[LeaderStorage] Using direct Redis URL: {self._override_url}") @@ -85,24 +103,108 @@ class LeaderStorage(StorageProvider): # Set up change callback self._leader_client.on_change(self._on_leader_change) - # Wait for leader and connect - 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) + # Wait for leader with retry loop + start_time = time.time() + attempt = 0 - # Configure WebDAV client with leader's WebDAV URL - if info.webdav_url: - webdav_client.configure(info.webdav_url) - except TimeoutError as e: - print(f"[LeaderStorage] Failed to find leader: {e}") - return + while True: + attempt += 1 + elapsed = time.time() - start_time - # Start watching for leader changes - self._leader_client.start_watching() + # Check timeout (0 = no timeout) + if LEADER_WAIT_TIMEOUT > 0 and elapsed > LEADER_WAIT_TIMEOUT: + print(f"[LeaderStorage] Timed out waiting for leader after {LEADER_WAIT_TIMEOUT}s") + # Start background reconnection thread + self._start_background_reconnection() + return + + try: + # Try to get leader info (short timeout per attempt) + info = self._leader_client.get_leader_info() + + if info and info.redis_url: + print(f"[LeaderStorage] Found leader at {info.redis_url} (attempt {attempt})") + self._connect_to_redis(info.redis_url) + + if self._connected: + # Configure WebDAV client with leader's internal WebDAV URL (for container-to-container access) + if info.webdav_url_internal: + webdav_client.configure(info.webdav_url_internal) + print(f"[LeaderStorage] WebDAV configured: {info.webdav_url_internal}") + elif info.webdav_url: + # Fallback to external URL if internal not available + webdav_client.configure(info.webdav_url) + print(f"[LeaderStorage] WebDAV configured (fallback): {info.webdav_url}") + + # Start watching for leader changes + self._leader_client.start_watching() + return + else: + print(f"[LeaderStorage] Failed to connect to Redis, retrying...") + else: + print(f"[LeaderStorage] Waiting for meta-core leader... (attempt {attempt})") + + except Exception as e: + print(f"[LeaderStorage] Error during leader discovery: {e}") + + # Wait before retry + time.sleep(LEADER_RETRY_INTERVAL) + + def _start_background_reconnection(self) -> None: + """Start a background thread that keeps trying to connect to the leader.""" + if hasattr(self, '_reconnect_thread') and self._reconnect_thread and self._reconnect_thread.is_alive(): + return # Already running + + print("[LeaderStorage] Starting background reconnection thread...") + self._reconnect_stop = threading.Event() + self._reconnect_thread = threading.Thread( + target=self._background_reconnect_loop, + daemon=True, + name="leader-reconnect" + ) + self._reconnect_thread.start() + + def _background_reconnect_loop(self) -> None: + """Background thread that keeps trying to connect to the leader.""" + attempt = 0 + + while not self._reconnect_stop.is_set(): + attempt += 1 + + try: + info = self._leader_client.get_leader_info() + + if info and info.redis_url: + print(f"[LeaderStorage] Background: Found leader at {info.redis_url}") + self._connect_to_redis(info.redis_url) + + if self._connected: + # Configure WebDAV client + if info.webdav_url_internal: + webdav_client.configure(info.webdav_url_internal) + elif info.webdav_url: + webdav_client.configure(info.webdav_url) + + # Start watching for leader changes + self._leader_client.start_watching() + print("[LeaderStorage] Background: Successfully connected to leader!") + return # Exit the reconnection loop + + except Exception as e: + print(f"[LeaderStorage] Background reconnect error: {e}") + + # Wait before retry + self._reconnect_stop.wait(LEADER_RETRY_INTERVAL) def disconnect(self) -> None: """Disconnect from the storage backend.""" + # Stop background reconnection thread if running + if hasattr(self, '_reconnect_stop') and self._reconnect_stop: + self._reconnect_stop.set() + if hasattr(self, '_reconnect_thread') and self._reconnect_thread: + self._reconnect_thread.join(timeout=2.0) + self._reconnect_thread = None + # Stop leader client if self._leader_client: self._leader_client.close() @@ -129,18 +231,34 @@ class LeaderStorage(StorageProvider): self._disconnect_redis() self._notify_disconnect() - # Reconnect to new leader + # Reconnect to new leader using retry loop 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) + attempt = 0 + max_attempts = 60 # Try for ~5 minutes with 5s interval - # Reconfigure WebDAV client with new leader's WebDAV URL - if info.webdav_url: - webdav_client.configure(info.webdav_url) - except TimeoutError as e: - print(f"[LeaderStorage] Failed to reconnect: {e}") + while attempt < max_attempts: + attempt += 1 + try: + info = self._leader_client.get_leader_info() + + if info and info.redis_url: + print(f"[LeaderStorage] Reconnecting to leader at {info.redis_url}...") + self._connect_to_redis(info.redis_url) + + if self._connected: + # Reconfigure WebDAV client with new leader's internal WebDAV URL + if info.webdav_url_internal: + webdav_client.configure(info.webdav_url_internal) + elif info.webdav_url: + # Fallback to external URL if internal not available + webdav_client.configure(info.webdav_url) + return + except Exception as e: + print(f"[LeaderStorage] Reconnect attempt {attempt} failed: {e}") + + time.sleep(LEADER_RETRY_INTERVAL) + + print(f"[LeaderStorage] Failed to reconnect after {max_attempts} attempts") def _connect_to_redis(self, url: str) -> None: """Connect to Redis.""" @@ -606,6 +724,7 @@ class LeaderStorage(StorageProvider): 'api_url': info.api_url, 'redis_url': info.redis_url, 'webdav_url': info.webdav_url, + 'webdav_url_internal': info.webdav_url_internal, } status = {