Merge pull request ManiMatter#366 from jrhager84/startup-degraded-mode

Degrade unreachable instances at startup instead of exiting the whole app

Author: jrhager84

main.py

@@ -8,7 +8,7 @@
 from src.job_manager import JobManager
 from src.settings.settings import Settings
 from src.utils.log_setup import logger
-from src.utils.startup import launch_steps
+from src.utils.startup import launch_steps, retry_degraded_instances
 
 settings = Settings()
 job_manager = JobManager(settings)
@@ -53,13 +53,18 @@ async def main():
     await launch_steps(settings)
 
     if settings.jobs.detect_deletions.enabled:
-        await WatcherManager(settings).setup()
+        await watch_manager.setup()
     # Start Cleaning
     while True:
         logger.info("-" * 50)
 
+        # Give degraded instances a chance to rejoin before this cycle's jobs
+        await retry_degraded_instances(settings, watch_manager)
+
         # Refresh qBit Cookies (SABnzbd doesn't need cookie refresh)
         for qbit in settings.download_clients.qbittorrent:
+            if not qbit.ready:
+                continue
             try:
                 await qbit.refresh_cookie()
             except Exception as err:  # noqa: BLE001
@@ -70,6 +75,8 @@ async def main():
 
         # Run script for each instance
         for arr in settings.instances:
+            if not arr.ready:  # skip was already logged by retry_degraded_instances
+                continue
             await job_manager.run_jobs(arr)
             logger.verbose("")
 

src/deletion_handler/deletion_handler.py

@@ -99,35 +99,52 @@ async def setup(self):
         for arr, folder_path in folders_to_watch:
             self.set_watcher(arr, folder_path)
 
+    async def setup_for_arr(self, arr):
+        """Set up deletion watchers for a single arr (e.g. one that rejoined after a degraded startup)."""
+        if self.loop is None:
+            self.loop = asyncio.get_running_loop()
+        for folder_path in await self.get_folders_to_watch_for_arr(arr):
+            self.set_watcher(arr, folder_path)
+
     async def get_folders_to_watch(self):
         """Gets from all arrs the root folders and lists those that are accessible for the arr, and have present for decluttarr."""
         folders_to_watch = []
         logger.verbose("")
         logger.verbose("*** Setting up monitoring for deletions ***")
         for arr in self.settings.instances:
-            if arr.arr_type not in (
-                "sonarr",
-                "radarr",
-            ):  # only working for sonarr / radarr for now
-                continue
-            root_folders = await arr.get_root_folders()
-
-            for folder in root_folders:
-                if folder.get("accessible") and "path" in folder:
-                    path = Path(folder["path"])
-                    if path.exists():
-                        folders_to_watch.append((arr, folder["path"]))
-                    else:
-                        logger.warning(
-                            f"Job 'detect_deletions' on {arr.name} ({arr.base_url}) does not have access to this path and will not monitor it: '{path}'"
-                        )
+            for folder_path in await self.get_folders_to_watch_for_arr(arr):
+                folders_to_watch.append((arr, folder_path))
+
+        return folders_to_watch
+
+    async def get_folders_to_watch_for_arr(self, arr):
+        """Root folder paths of one arr that are accessible for both the arr and decluttarr."""
+        folders_to_watch = []
+        if arr.arr_type not in (
+            "sonarr",
+            "radarr",
+        ):  # only working for sonarr / radarr for now
+            return folders_to_watch
+        if not arr.ready:  # degraded instance; watchers are added when it rejoins
+            return folders_to_watch
+        root_folders = await arr.get_root_folders()
+
+        for folder in root_folders:
+            if folder.get("accessible") and "path" in folder:
+                path = Path(folder["path"])
+                if path.exists():
+                    folders_to_watch.append(folder["path"])
+                else:
+                    logger.warning(
+                        f"Job 'detect_deletions' on {arr.name} ({arr.base_url}) does not have access to this path and will not monitor it: '{path}'"
+                    )
+                    logger.info(
+                        ">>> 💡 Tip: Make sure that the paths in decluttarr and in your arr instance are identical."
+                    )
+                    if self.settings.envs.in_docker:
                         logger.info(
-                            ">>> 💡 Tip: Make sure that the paths in decluttarr and in your arr instance are identical."
+                            ">>> 💡 Tip: Make sure decluttarr and your arr instance have the same mount points"
                         )
-                        if self.settings.envs.in_docker:
-                            logger.info(
-                                ">>> 💡 Tip: Make sure decluttarr and your arr instance have the same mount points"
-                            )
 
         return folders_to_watch
 

src/job_manager.py

@@ -56,6 +56,9 @@ async def run_download_client_jobs(self):
             )
 
             for client in download_clients:
+                if not client.ready:
+                    continue
+
                 # Get jobs for this client
                 download_client_jobs = self._get_download_client_jobs_for_client(
                     client,
@@ -150,6 +153,8 @@ async def _download_clients_connected(self):
 
     async def _check_client_connection_status(self, clients):
         for client in clients:
+            if not client.ready:  # never-set-up client must not veto or be polled
+                continue
             logger.debug(
                 f"job_manager.py/_check_client_connection_status: Checking if {client.name} is connected",
             )

src/jobs/removal_handler.py

@@ -44,6 +44,8 @@ async def _tag_as_obsolete(self, affected_download, download_id):
             f"Job '{self.job_name}' triggered obsolete-tagging: {affected_download['title']}"
         )
         for qbit in self.settings.download_clients.qbittorrent:
+            if not qbit.ready:
+                continue
             await qbit.set_tag(
                 tags=[self.settings.general.obsolete_tag], hashes=[download_id]
             )
@@ -55,12 +57,12 @@ async def _get_handling_method(self, download_id, affected_download):
         download_client_name = affected_download["downloadClient"]
         _, download_client_type = (
             self.settings.download_clients.get_download_client_by_name(
-                download_client_name
+                download_client_name, ready_only=True
             )
         )
 
         if download_client_type != "qbittorrent":
-            return "remove"  # handling is only implemented for qbit
+            return "remove"  # handling is only implemented for qbit (and only if ready)
 
         if len(self.settings.download_clients.qbittorrent) == 0:
             return "remove"  # qbit not configured, thus can't tag

src/jobs/removal_job.py

@@ -50,6 +50,7 @@ async def run(self) -> int:
 
         # -- Checks --
         self._ignore_protected()
+        self._ignore_degraded_client_downloads()
         if self.max_strikes:
             self.affected_downloads = self.strikes_handler.filter_strike_exceeds(
                 self.affected_downloads, self.queue
@@ -76,6 +77,36 @@ def _ignore_protected(self):
             if download_id not in self.arr.tracker.protected
         }
 
+    def _ignore_degraded_client_downloads(self):
+        """Fail-closed: never remove a download whose configured client is degraded.
+
+        A degraded (not-ready) download client can't be queried for protection
+        status (protected tag, private/public tracker), so its downloads must be
+        left untouched rather than deleted blindly. Downloads on healthy clients,
+        or on clients not configured in decluttarr, are unaffected.
+        """
+        safe = {}
+        skipped = 0
+        for download_id, download in self.affected_downloads.items():
+            # download is the grouped metadata dict from group_by_download_id.
+            client_name = download.get("downloadClient")
+            client = None
+            if client_name:
+                client, _ = self.settings.download_clients.get_download_client_by_name(
+                    client_name
+                )
+            if client is not None and not client.ready:
+                skipped += 1
+                continue
+            safe[download_id] = download
+
+        if skipped:
+            logger.warning(
+                f">>> {self.job_name}: skipped {skipped} download(s) on {self.arr.name} "
+                "whose download client is degraded (protection can't be verified; left untouched).",
+            )
+        self.affected_downloads = safe
+
     @abstractmethod  # Implemented on level of each removal job
     async def _find_affected_items(self) -> None:
         pass

src/jobs/remove_bad_files.py

@@ -62,7 +62,7 @@ def _group_download_ids_by_client(self):
 
             download_client, download_client_type = (
                 self.settings.download_clients.get_download_client_by_name(
-                    download_client_name
+                    download_client_name, ready_only=True
                 )
             )
             if not download_client or not download_client_type:

src/jobs/remove_slow.py

@@ -168,7 +168,7 @@ async def add_download_client_to_queue_items(self):
             download_client_name = item["downloadClient"]
             download_client, download_client_type = (
                 self.settings.download_clients.get_download_client_by_name(
-                    download_client_name
+                    download_client_name, ready_only=True
                 )
             )
             item["download_client"] = download_client

src/settings/_download_clients.py

@@ -81,11 +81,16 @@ def check_unique_download_client_types(self):
                 seen.add(name.lower())
 
     def get_download_client_by_name(
-        self, name: str, download_client_type: str | None = None
+        self,
+        name: str,
+        download_client_type: str | None = None,
+        ready_only: bool = False,
     ):
         """
         Retrieve the download client and download client type by its name.
         If download_client_type is provided, search only in that type.
+        If ready_only is True, a client that failed its startup check (not ready)
+        is treated as unavailable, so degraded clients never get called from jobs.
         """
         name_lower = name.lower()
         types_to_search = (
@@ -97,6 +102,8 @@ def get_download_client_by_name(
 
             for download_client in download_clients:
                 if download_client.name.lower() == name_lower:
+                    if ready_only and not getattr(download_client, "ready", False):
+                        return None, None
                     return download_client, client_type
 
         return None, None

src/settings/_download_clients_qbit.py

@@ -2,12 +2,19 @@
 from requests.cookies import RequestsCookieJar
 
 from src.settings._constants import ApiEndpoints, MinVersions
-from src.utils.common import extract_json_from_response, make_request, wait_and_exit
+from src.utils.common import (
+    extract_json_from_response,
+    is_definitive_setup_error,
+    make_request,
+)
 from src.utils.log_setup import logger
 
 
 class QbitError(Exception):
-    pass
+    def __init__(self, message, tip="", definitive=False):
+        super().__init__(message)
+        self.tip = tip
+        self.definitive = definitive
 
 
 class QbitClients(list):
@@ -39,6 +46,10 @@ class QbitClient:
     cookie: dict[str, str] = None
     version: str = None
     bandwidth_usage: int = 0
+    ready: bool = False
+    failure_kind: str = None  # None | "transient" | "definitive"
+    last_error: str = None
+    setup_tip: str = ""
 
     def __init__(
         self,
@@ -160,7 +171,7 @@ async def validate_version(self):
                 f"Please update qBittorrent to at least version {min_version}. Current version: {self.version}",
             )
             error = f"qBittorrent version {self.version} is too old. Please update."
-            raise QbitError(error)
+            raise QbitError(error, definitive=True)
         if version.parse(self.version) < version.parse("5.0.0"):
             logger.info(
                 "[Tip!] Consider upgrading to qBittorrent v5.0.0 or newer to reduce network overhead.",
@@ -228,7 +239,7 @@ async def set_unwanted_folder(self):
                 )
 
     async def check_qbit_reachability(self):
-        """Check if the qBittorrent URL is reachable."""
+        """Check if the qBittorrent URL is reachable (and the credentials work)."""
         try:
             logger.debug(
                 "_download_clients_qBit.py/check_qbit_reachability: Checking if qbit is reachable",
@@ -239,7 +250,7 @@ async def check_qbit_reachability(self):
                 "password": getattr(self, "password", ""),
             }
             headers = {"content-type": "application/x-www-form-urlencoded"}
-            await make_request(
+            response = await make_request(
                 "post",
                 endpoint,
                 self.settings,
@@ -249,11 +260,21 @@ async def check_qbit_reachability(self):
                 log_error=False,
                 ignore_test_run=True,
             )
-
         except Exception as e:  # noqa: BLE001
             tip = "💡 Tip: Did you specify the URL (and username/password if required) correctly?"
-            logger.error(f"-- | qBittorrent\n❗️ {e}\n{tip}\n")
-            wait_and_exit()
+            if str(e) != self.last_error:  # Only report new failure modes in full
+                logger.error(f"-- | qBittorrent\n❗️ {e}\n{tip}\n")
+            raise QbitError(e, tip=tip) from e
+
+        # Bad credentials: qBit answers HTTP 200 with the body "Fails." Treat this
+        # as a definitive config error so we don't retry-loop every cycle and get
+        # the source IP banned by qBittorrent's failed-login protection.
+        if getattr(response, "text", None) == "Fails.":
+            tip = "💡 Tip: Check the qBittorrent username/password."
+            error = "qBittorrent login failed (incorrect username/password)."
+            if error != self.last_error:
+                logger.error(f"-- | qBittorrent\n❗️ {error}\n{tip}\n")
+            raise QbitError(error, tip=tip, definitive=True)
 
     async def check_connected(self):
         """Check if the qBittorrent is connected to internet."""
@@ -283,26 +304,38 @@ async def check_connected(self):
         return True
 
     async def setup(self):
-        """Perform the qBittorrent setup by calling relevant managers."""
-        # Check reachabilty
-        await self.check_qbit_reachability()
+        """Perform the qBittorrent setup; degrade instead of exiting on failure."""
+        try:
+            await self.check_qbit_reachability()
 
-        # Refresh the qBittorrent cookie first
-        await self.refresh_cookie()
+            # Refresh the qBittorrent cookie first
+            await self.refresh_cookie()
 
-        try:
             # Fetch version and validate it
             await self.fetch_version()
             await self.validate_version()
-            logger.info(f"OK | qBittorrent ({self.base_url})")
-        except QbitError as e:
-            logger.error(f"qBittorrent version check failed: {e}")
-            wait_and_exit()  # Exit if version check fails
 
-        # Continue with other setup tasks regardless of version check result
-        await self.create_required_tags()
-        await self.set_unwanted_folder()
-        await self.warn_no_bandwidth_limit_set()
+            await self.create_required_tags()
+            await self.set_unwanted_folder()
+            await self.warn_no_bandwidth_limit_set()
+
+            logger.info(f"OK | qBittorrent ({self.base_url})")
+            self.ready = True
+            self.failure_kind = None
+            self.last_error = None
+            self.setup_tip = ""
+        except Exception as e:  # noqa: BLE001
+            if not isinstance(e, QbitError) and str(e) != self.last_error:
+                logger.error(
+                    f"Unhandled error during qBittorrent setup: {e}", exc_info=True
+                )
+            self.ready = False
+            self.failure_kind = (
+                "definitive" if is_definitive_setup_error(e) else "transient"
+            )
+            self.last_error = str(e)
+            self.setup_tip = getattr(e, "tip", "")
+        return self.ready
 
     async def get_protected_and_private(self):
         """Fetch torrents from qBittorrent and checks for protected and private status."""
@@ -429,7 +462,6 @@ async def get_torrent_properties(self, qbit_hash):
         )
         return response.json()
 
-
     async def get_torrent_files(self, download_id):
         # this may not work if the wrong qbit
         logger.debug("_download_clients_qBit/get_torrent_files: Getting torrent files")