feat(startup): add dependency-aware parallel startup and shutdown-safe waits

- run service startup in parallel while honoring deps (postgres→riven/zilean, riven→frontend, pgadmin→postgres)
- add mount/url wait logic to process start and make waits shutdown-safe
- ensure rclone waits on decypharr/nzbdav webdav endpoints
- inject mount wait lists for media/arr services
- addresses potential shutdown issues during dependency checks by
incorporating shutdown checks.
- updates Zilean and Riven connection strings based on the
Postgres configuration.

Author: I-am-PUID-0

main.py

@@ -8,7 +8,8 @@
 from utils.auto_update import Update
 from utils.dependencies import initialize_dependencies
 from utils.plex_dbrepair import start_plex_dbrepair_worker
-import subprocess, threading, time, tomllib, os, socket, errno, psutil
+from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
+import subprocess, threading, time, tomllib, os, socket, errno, psutil, json
 
 
 def log_ascii_art():
@@ -265,6 +266,153 @@ def start_configured_process(config_obj, updater, key_name, exit_on_error=True):
             raise
 
 
+def _service_has_enabled_instance(config_obj: dict) -> bool:
+    if not isinstance(config_obj, dict):
+        return False
+    if "instances" in config_obj and isinstance(config_obj["instances"], dict):
+        return any(
+            isinstance(inst, dict) and inst.get("enabled")
+            for inst in config_obj["instances"].values()
+        )
+    return bool(config_obj.get("enabled"))
+
+
+def _read_decypharr_mount_path(decypharr_cfg: dict) -> str | None:
+    if not decypharr_cfg.get("use_embedded_rclone"):
+        return None
+    config_file = decypharr_cfg.get("config_file")
+    if config_file and os.path.exists(config_file):
+        try:
+            with open(config_file, "r") as handle:
+                data = json.load(handle)
+            mount_path = (data.get("rclone") or {}).get("mount_path")
+            if isinstance(mount_path, str) and mount_path.strip():
+                return mount_path
+        except Exception as e:
+            logger.debug("Failed to read Decypharr mount path: %s", e)
+    return "/mnt/debrid/decypharr"
+
+
+def _collect_mount_paths(config_manager) -> list[str]:
+    mount_paths = set()
+    rclone_instances = config_manager.get("rclone", {}).get("instances", {}) or {}
+    for instance in rclone_instances.values():
+        if not isinstance(instance, dict) or not instance.get("enabled"):
+            continue
+        mount_dir = instance.get("mount_dir")
+        mount_name = instance.get("mount_name")
+        if mount_dir and mount_name:
+            mount_paths.add(os.path.join(mount_dir, mount_name))
+
+    decypharr_cfg = config_manager.get("decypharr", {}) or {}
+    if decypharr_cfg.get("enabled") and decypharr_cfg.get("use_embedded_rclone"):
+        mount_path = _read_decypharr_mount_path(decypharr_cfg)
+        if mount_path:
+            mount_paths.add(mount_path)
+
+    return sorted(mount_paths)
+
+
+def _merge_wait_for_mounts(config_obj: dict, mount_paths: list[str]) -> None:
+    existing = config_obj.get("wait_for_mounts") or []
+    merged = sorted(set(existing) | set(mount_paths))
+    if merged:
+        config_obj["wait_for_mounts"] = merged
+
+
+def _apply_mount_waits(config_manager, mount_paths: list[str]) -> None:
+    if not mount_paths:
+        return
+    mount_wait_keys = {
+        "plex",
+        "jellyfin",
+        "emby",
+    }
+    for key in mount_wait_keys:
+        cfg = config_manager.get(key, {})
+        if not isinstance(cfg, dict):
+            continue
+        if "instances" in cfg and isinstance(cfg["instances"], dict):
+            for inst in cfg["instances"].values():
+                if isinstance(inst, dict) and inst.get("enabled"):
+                    _merge_wait_for_mounts(inst, mount_paths)
+        elif cfg.get("enabled"):
+            _merge_wait_for_mounts(cfg, mount_paths)
+
+
+def _build_dependency_map(config_manager) -> dict[str, set[str]]:
+    deps = {
+        "riven_backend": {"postgres"},
+        "riven_frontend": {"riven_backend"},
+        "zilean": {"postgres"},
+        "pgadmin": {"postgres"},
+    }
+
+    rclone_deps = set()
+    rclone_instances = config_manager.get("rclone", {}).get("instances", {}) or {}
+    for instance in rclone_instances.values():
+        if not isinstance(instance, dict) or not instance.get("enabled"):
+            continue
+        if instance.get("zurg_enabled"):
+            rclone_deps.add("zurg")
+        if instance.get("decypharr_enabled"):
+            rclone_deps.add("decypharr")
+        key_type = (instance.get("key_type") or "").lower()
+        if key_type == "nzbdav" or instance.get("core_service") == "nzbdav":
+            rclone_deps.add("nzbdav")
+    if rclone_deps:
+        deps["rclone"] = rclone_deps
+
+    return deps
+
+
+def _start_processes_with_dependencies(
+    process_handler, updater, config_manager, keys: list[str], dependency_map
+) -> None:
+    enabled = {
+        key: _service_has_enabled_instance(config_manager.get(key, {})) for key in keys
+    }
+    pending = {key for key in keys if enabled.get(key)}
+    deps = {
+        key: {d for d in dependency_map.get(key, set()) if enabled.get(d)}
+        for key in pending
+    }
+
+    def _start_key(key: str) -> None:
+        cfg = config_manager.get(key, {})
+        start_configured_process(cfg, updater, key)
+
+    in_progress = {}
+    completed = set()
+    with ThreadPoolExecutor() as executor:
+        while pending or in_progress:
+            if process_handler.shutting_down:
+                for future in list(in_progress):
+                    future.cancel()
+                return
+            ready = [key for key in list(pending) if deps.get(key, set()) <= completed]
+            for key in ready:
+                if process_handler.shutting_down:
+                    break
+                pending.remove(key)
+                in_progress[executor.submit(_start_key, key)] = key
+
+            if not in_progress:
+                raise RuntimeError(
+                    f"Dependency resolution stalled. Remaining: {sorted(pending)}"
+                )
+
+            done, _ = wait(in_progress.keys(), return_when=FIRST_COMPLETED)
+            for future in done:
+                key = in_progress.pop(future)
+                try:
+                    future.result()
+                except Exception as e:
+                    logger.error("Failed while starting %s: %s", key, e)
+                    process_handler.shutdown(exit_code=1)
+                completed.add(key)
+
+
 def main():
     log_ascii_art()
 
@@ -290,6 +438,8 @@ def main():
         process_handler.shutdown(exit_code=1)
 
     _apply_global_port_reservations(config)
+    mount_paths = _collect_mount_paths(config)
+    _apply_mount_waits(config, mount_paths)
 
     if config.get("dumb", {}).get("api_service", {}).get("enabled"):
         start_fastapi_process()
@@ -351,8 +501,10 @@ def _get_metrics_cfg():
             time.sleep(interval)
 
     try:
+        if config.get("traefik", {}).get("enabled"):
+            start_configured_process(config.get("traefik", {}), updater, "traefik")
+
         grouped_keys = [
-            "traefik",
             "zurg",
             "prowlarr",
             "radarr",
@@ -377,9 +529,10 @@ def _get_metrics_cfg():
             "tautulli",
             "seerr",
         ]
-        for key in grouped_keys:
-            cfg = config.get(key, {})
-            start_configured_process(cfg, updater, key)
+        dependency_map = _build_dependency_map(config)
+        _start_processes_with_dependencies(
+            process_handler, updater, config, grouped_keys, dependency_map
+        )
 
     except Exception as e:
         logger.error(e)

utils/auto_update.py

@@ -507,11 +507,40 @@ def start_process(self, process_name, config, key, instance_name):
 
         if config.get("wait_for_dir", False):
             while not os.path.exists(wait_dir := config["wait_for_dir"]):
+                if self.process_handler.shutting_down:
+                    self.logger.info(
+                        "Shutdown requested; skipping wait for directory %s.",
+                        wait_dir,
+                    )
+                    return False, "Shutdown requested"
                 self.logger.info(
                     f"Waiting for directory {wait_dir} to become available before starting {process_name}"
                 )
                 time.sleep(10)
 
+        wait_mounts = config.get("wait_for_mounts") or []
+        if wait_mounts:
+            while True:
+                if self.process_handler.shutting_down:
+                    self.logger.info(
+                        "Shutdown requested; skipping wait for mounts before %s.",
+                        process_name,
+                    )
+                    return False, "Shutdown requested"
+                missing = [
+                    mount_path
+                    for mount_path in wait_mounts
+                    if not os.path.ismount(mount_path)
+                ]
+                if not missing:
+                    break
+                self.logger.info(
+                    "Waiting for mounts to become available before starting %s: %s",
+                    process_name,
+                    ", ".join(missing),
+                )
+                time.sleep(10)
+
         if config.get("wait_for_url", False):
             wait_for_urls = config["wait_for_url"]
             time.sleep(5)
@@ -526,6 +555,12 @@ def start_process(self, process_name, config, key, instance_name):
                 )
 
                 while time.time() - start_time < 600:
+                    if self.process_handler.shutting_down:
+                        self.logger.info(
+                            "Shutdown requested; skipping wait for %s.",
+                            wait_url,
+                        )
+                        return False, "Shutdown requested"
                     try:
                         if auth:
                             response = requests.get(
@@ -581,13 +616,17 @@ def start_process(self, process_name, config, key, instance_name):
             suppress_logging=suppress_logging,
             env=env,
         )
+        if self.process_handler.shutting_down:
+            return process, "Shutdown requested"
         if key == "riven_backend":
             from utils.riven_settings import load_settings
 
             time.sleep(10)
             load_settings()
 
         if key == "decypharr":
+            if self.process_handler.shutting_down:
+                return process, "Shutdown requested"
             from utils.decypharr_settings import patch_decypharr_config
 
             time.sleep(10)
@@ -607,6 +646,8 @@ def start_process(self, process_name, config, key, instance_name):
                 self.logger.warning("Decypharr config patch failed: %s", error)
 
         if key == "nzbdav":
+            if self.process_handler.shutting_down:
+                return process, "Shutdown requested"
             from utils.nzbdav_settings import patch_nzbdav_config
 
             time.sleep(10)
@@ -634,6 +675,8 @@ def start_process(self, process_name, config, key, instance_name):
             "whisparr",
             "whisparr-v3",
         ]:
+            if self.process_handler.shutting_down:
+                return process, "Shutdown requested"
             from utils.prowlarr_settings import patch_prowlarr_apps
 
             time.sleep(10)
@@ -642,6 +685,8 @@ def start_process(self, process_name, config, key, instance_name):
                 self.logger.warning("Prowlarr app sync failed: %s", err)
 
         if key == "plex":
+            if self.process_handler.shutting_down:
+                return process, "Shutdown requested"
             from utils.plex_settings import patch_plex_config
 
             time.sleep(10)

utils/decypharr_settings.py

@@ -364,6 +364,8 @@ def _wait_for_arr(
     deadline = time.time() + max(1, timeout_s)
     url = _join(host, "/api/v3/system/status")
     while time.time() < deadline:
+        if _shutdown_requested():
+            return False
         try:
             _arr_req(url, token, "GET", timeout=5)
             return True
@@ -384,6 +386,8 @@ def _with_retries(
     last_err = None
     delay = max(0.1, base_delay_s)
     for i in range(max(1, attempts)):
+        if _shutdown_requested():
+            raise RuntimeError("Shutdown requested")
         try:
             return fn(*args, **kwargs)
         except Exception as e:
@@ -395,12 +399,24 @@ def _with_retries(
     raise last_err if last_err else RuntimeError("retry failed")
 
 
+def _shutdown_requested() -> bool:
+    try:
+        from utils.dependencies import get_process_handler
+
+        handler = get_process_handler()
+    except Exception:
+        return False
+    return bool(getattr(handler, "shutting_down", False))
+
+
 # ---------------------------------------------------------------------------
 # Main patch entrypoint
 # ---------------------------------------------------------------------------
 
 
 def patch_decypharr_config():
+    if _shutdown_requested():
+        return False, "Shutdown requested"
     config_path = CONFIG_MANAGER.get("decypharr", {}).get(
         "config_file", "/decypharr/config.json"
     )