feat(setup): implement project setup with preinstall support and locking mechanism

Adds a preinstall feature to service setup, allowing for preparatory installations before service startup. This improves initial startup times and dependency handling.

Also introduces an apt locking mechanism to prevent conflicts during package management operations, ensuring smoother installations and updates.

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
-from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
+from utils.setup import setup_project
+from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED, as_completed
 import subprocess, threading, time, tomllib, os, socket, errno, psutil, json, urllib.parse
 
 
@@ -541,6 +542,58 @@ def _apply_waits_to_service(config_manager, key: str, wait_entries: list[dict])
         _set_wait_for_urls(cfg, wait_entries)
 
 
+def _collect_enabled_process_names(config_manager) -> list[str]:
+    process_names = []
+    for key, cfg in config_manager.config.items():
+        if not isinstance(cfg, dict):
+            continue
+        if key == "dumb":
+            continue
+        if "instances" in cfg and isinstance(cfg["instances"], dict):
+            for inst_cfg in cfg["instances"].values():
+                if isinstance(inst_cfg, dict) and inst_cfg.get("enabled"):
+                    process_name = inst_cfg.get("process_name")
+                    if process_name:
+                        process_names.append(process_name)
+            continue
+        if cfg.get("enabled"):
+            process_name = cfg.get("process_name")
+            if process_name:
+                process_names.append(process_name)
+    return process_names
+
+
+def _preinstall_enabled_services(process_handler, config_manager) -> None:
+    process_names = _collect_enabled_process_names(config_manager)
+    if not process_names:
+        return
+    logger.info("Pre-installing enabled services before startup.")
+    max_workers = min(4, max(1, len(process_names)))
+
+    def _run_preinstall(name: str) -> None:
+        if process_handler.shutting_down:
+            return
+        with process_handler.process_context(name):
+            key, _ = config_manager.find_key_for_process(name)
+            if key in {"pgadmin", "postgres"}:
+                logger.info("Preinstall skip for %s; requires running dependency.", name)
+                return
+            success, error = setup_project(process_handler, name, preinstall=True)
+            if not success:
+                raise RuntimeError(error)
+
+    with ThreadPoolExecutor(max_workers=max_workers) as executor:
+        futures = {executor.submit(_run_preinstall, name): name for name in process_names}
+        for future in as_completed(futures):
+            name = futures[future]
+            try:
+                future.result()
+            except Exception as e:
+                logger.error("Pre-install failed for %s: %s", name, e)
+                process_handler.shutdown(exit_code=1)
+                raise
+
+
 def _build_dependency_map(config_manager) -> dict[str, set[str]]:
     deps = {
         "riven_backend": {"postgres"},
@@ -676,6 +729,7 @@ def main():
     _apply_prowlarr_waits(config, _collect_arr_ping_waits(config))
     _apply_waits_to_service(config, "tautulli", _build_plex_wait_entries(config))
     _apply_waits_to_service(config, "seerr", _build_media_wait_entries(config))
+    _preinstall_enabled_services(process_handler, config)
 
     if config.get("dumb", {}).get("api_service", {}).get("enabled"):
         start_fastapi_process()

utils/apt_lock.py

@@ -0,0 +1,52 @@
+from contextlib import contextmanager
+import threading
+import time
+import subprocess
+
+from utils.global_logger import logger
+
+
+_APT_LOCK = threading.Lock()
+
+
+def _looks_like_lock_error(stderr: str) -> bool:
+    if not stderr:
+        return False
+    needle = stderr.lower()
+    return (
+        "dpkg frontend lock" in needle
+        or "could not get lock" in needle
+        or "unable to acquire the dpkg frontend lock" in needle
+        or "could not acquire dpkg frontend lock" in needle
+    )
+
+
+@contextmanager
+def apt_lock():
+    _APT_LOCK.acquire()
+    try:
+        yield
+    finally:
+        _APT_LOCK.release()
+
+
+def run_locked(cmd, *, retries: int = 6, delay_s: float = 5.0, **kwargs):
+    with apt_lock():
+        last_err = None
+        for attempt in range(max(1, retries)):
+            try:
+                return subprocess.run(cmd, **kwargs)
+            except subprocess.CalledProcessError as e:
+                stderr = getattr(e, "stderr", "") or ""
+                if not _looks_like_lock_error(stderr) or attempt == retries - 1:
+                    raise
+                wait_s = delay_s * (attempt + 1)
+                logger.warning(
+                    "dpkg/apt lock busy; retrying in %.1fs (%s)",
+                    wait_s,
+                    cmd,
+                )
+                time.sleep(wait_s)
+                last_err = e
+        if last_err:
+            raise last_err

utils/auto_update.py

@@ -203,8 +203,9 @@ def update_check(self, process_name, config, key, instance_name):
             )
             if process_name in self.process_handler.process_names:
                 self.stop_process(process_name)
-            if process_name in self.process_handler.setup_tracker:
-                self.process_handler.setup_tracker.remove(process_name)
+            with self.process_handler.setup_tracker_lock:
+                if process_name in self.process_handler.setup_tracker:
+                    self.process_handler.setup_tracker.remove(process_name)
             release_version = f"{update_info.get('latest_version')}"
             if not prerelease and not nightly:
                 config["release_version"] = release_version
@@ -256,8 +257,9 @@ def update_check_pinned_version(
         )
         if process_name in self.process_handler.process_names:
             self.stop_process(process_name)
-        if process_name in self.process_handler.setup_tracker:
-            self.process_handler.setup_tracker.remove(process_name)
+        with self.process_handler.setup_tracker_lock:
+            if process_name in self.process_handler.setup_tracker:
+                self.process_handler.setup_tracker.remove(process_name)
 
         success, error = setup_project(self.process_handler, process_name)
         if not success:
@@ -295,8 +297,9 @@ def update_check_jellyfin_latest(self, process_name, config, key, instance_name)
         )
         if process_name in self.process_handler.process_names:
             self.stop_process(process_name)
-        if process_name in self.process_handler.setup_tracker:
-            self.process_handler.setup_tracker.remove(process_name)
+        with self.process_handler.setup_tracker_lock:
+            if process_name in self.process_handler.setup_tracker:
+                self.process_handler.setup_tracker.remove(process_name)
 
         installer = JellyfinInstaller()
         success, error = installer.install_jellyfin_server()
@@ -335,8 +338,9 @@ def update_check_emby_latest(self, process_name, config, key, instance_name):
         )
         if process_name in self.process_handler.process_names:
             self.stop_process(process_name)
-        if process_name in self.process_handler.setup_tracker:
-            self.process_handler.setup_tracker.remove(process_name)
+        with self.process_handler.setup_tracker_lock:
+            if process_name in self.process_handler.setup_tracker:
+                self.process_handler.setup_tracker.remove(process_name)
 
         original_release_enabled = config.get("release_version_enabled")
         original_release_version = config.get("release_version")
@@ -376,8 +380,9 @@ def update_check_arr_latest(self, process_name, config, key, instance_name):
         )
         if process_name in self.process_handler.process_names:
             self.stop_process(process_name)
-        if process_name in self.process_handler.setup_tracker:
-            self.process_handler.setup_tracker.remove(process_name)
+        with self.process_handler.setup_tracker_lock:
+            if process_name in self.process_handler.setup_tracker:
+                self.process_handler.setup_tracker.remove(process_name)
 
         success, error = installer.install()
         if not success:
@@ -479,8 +484,9 @@ def update_check_plex(self, process_name, config, key, instance_name):
         )
         if process_name in self.process_handler.process_names:
             self.stop_process(process_name)
-        if process_name in self.process_handler.setup_tracker:
-            self.process_handler.setup_tracker.remove(process_name)
+        with self.process_handler.setup_tracker_lock:
+            if process_name in self.process_handler.setup_tracker:
+                self.process_handler.setup_tracker.remove(process_name)
 
         success, error = installer.install_plex_media_server()
         if not success:
@@ -512,6 +518,7 @@ def start_process(self, process_name, config, key, instance_name):
             instance_name = refreshed_instance
 
         if config.get("wait_for_dir", False):
+            sleep_s = 10
             while not os.path.exists(wait_dir := config["wait_for_dir"]):
                 if self.process_handler.shutting_down:
                     self.logger.info(
@@ -522,10 +529,12 @@ def start_process(self, process_name, config, key, instance_name):
                 self.logger.info(
                     f"Waiting for directory {wait_dir} to become available before starting {process_name}"
                 )
-                time.sleep(10)
+                time.sleep(sleep_s)
+                sleep_s = min(60, int(sleep_s * 1.5))
 
         wait_mounts = config.get("wait_for_mounts") or []
         if wait_mounts:
+            sleep_s = 10
             while True:
                 if self.process_handler.shutting_down:
                     self.logger.info(
@@ -545,7 +554,8 @@ def start_process(self, process_name, config, key, instance_name):
                     process_name,
                     ", ".join(missing),
                 )
-                time.sleep(10)
+                time.sleep(sleep_s)
+                sleep_s = min(60, int(sleep_s * 1.5))
 
         if config.get("wait_for_url", False):
             wait_for_urls = config["wait_for_url"]
@@ -560,6 +570,7 @@ def start_process(self, process_name, config, key, instance_name):
                     f"Waiting to start {process_name} until {wait_url} is accessible."
                 )
 
+                sleep_s = 5
                 while time.time() - start_time < 600:
                     if self.process_handler.shutting_down:
                         self.logger.info(
@@ -589,7 +600,8 @@ def start_process(self, process_name, config, key, instance_name):
                             )
                     except requests.RequestException as e:
                         logger.debug(f"Waiting for {wait_url}: {e}")
-                    time.sleep(5)
+                    time.sleep(sleep_s)
+                    sleep_s = min(60, int(sleep_s * 1.5))
                 else:
                     raise RuntimeError(
                         f"Timeout: {wait_url} is not accessible after 600 seconds."

utils/jellyfin.py

@@ -1,4 +1,5 @@
 from utils.global_logger import logger
+from utils.apt_lock import run_locked, apt_lock
 import subprocess, os
 
 
@@ -32,18 +33,19 @@ def install_jellyfin_server(self, version=None):
                 self.logger.info("Installing Jellyfin media server...")
 
             # Step 1: Ensure required tools
-            subprocess.run(["apt", "update"], check=True)
-            subprocess.run(["apt", "install", "-y", "gnupg", "curl"], check=True)
+            run_locked(["apt", "update"], check=True)
+            run_locked(["apt", "install", "-y", "gnupg", "curl"], check=True)
 
             # Step 2: Add universe repo
             with open("/etc/os-release") as f:
                 os_release = f.read().lower()
             if "ubuntu" in os_release:
-                subprocess.run(["add-apt-repository", "-y", "universe"], check=True)
+                run_locked(["add-apt-repository", "-y", "universe"], check=True)
 
             # Step 3: Create keyring and download GPG key
             os.makedirs("/etc/apt/keyrings", exist_ok=True)
-            self.download_and_install_jellyfin_gpg_key()
+            with apt_lock():
+                self.download_and_install_jellyfin_gpg_key()
 
             # Step 4: Create sources file
             version_os = subprocess.check_output(
@@ -72,15 +74,15 @@ def install_jellyfin_server(self, version=None):
                 f.write(sources_content)
 
             # Step 5: Update apt
-            subprocess.run(["apt", "update"], check=True)
+            run_locked(["apt", "update"], check=True)
 
             # Step 6: Install jellyfin metapackage
             if version:
-                subprocess.run(
+                run_locked(
                     ["apt", "install", "-y", f"jellyfin={version}"], check=True
                 )
             else:
-                subprocess.run(["apt", "install", "-y", "jellyfin"], check=True)
+                run_locked(["apt", "install", "-y", "jellyfin"], check=True)
 
             # Step 7: Ensure web client is available
             expected_web_path = "/usr/lib/jellyfin/bin/jellyfin-web"

utils/plex.py

@@ -1,4 +1,5 @@
 from utils.global_logger import logger
+from utils.apt_lock import run_locked
 from utils.config_loader import CONFIG_MANAGER
 from utils.versions import Versions
 import os, platform, subprocess, tempfile, requests
@@ -98,7 +99,7 @@ def install_plex_media_server(self, version=None):
                     for chunk in r.iter_content(chunk_size=8192):
                         tmp_file.write(chunk)
 
-            subprocess.run(
+            run_locked(
                 [
                     "dpkg",
                     "-i",

utils/postgres.py

@@ -620,7 +620,8 @@ def pgadmin_setup(process_handler):
 
         pgadmin_db_uri = f"postgresql://{postgres_user}:{postgres_password}@{postgres_host}:{postgres_port}/pgadmin"
 
-        process_handler.setup_tracker.add(pgadmin_process_name)
+        with process_handler.setup_tracker_lock:
+            process_handler.setup_tracker.add(pgadmin_process_name)
         success, error = start_pgadmin(
             process_handler,
             pgadmin_config_dir,
@@ -633,7 +634,8 @@ def pgadmin_setup(process_handler):
             pgadmin_default_server,
         )
         if not success:
-            process_handler.setup_tracker.remove(pgadmin_process_name)
+            with process_handler.setup_tracker_lock:
+                process_handler.setup_tracker.remove(pgadmin_process_name)
             return False, error
 
         server_details = {
@@ -647,7 +649,8 @@ def pgadmin_setup(process_handler):
 
         success, error = add_pgadmin_server_to_db(pgadmin_db_uri, server_details)
         if not success:
-            process_handler.setup_tracker.remove(pgadmin_process_name)
+            with process_handler.setup_tracker_lock:
+                process_handler.setup_tracker.remove(pgadmin_process_name)
             return False, error
 
         success, error = start_pgagent(
@@ -657,11 +660,13 @@ def pgadmin_setup(process_handler):
             return False, error
 
         logger.info("pgAdmin setup completed successfully.")
-        process_handler.setup_tracker.add(pgadmin_process_name)
+        with process_handler.setup_tracker_lock:
+            process_handler.setup_tracker.add(pgadmin_process_name)
         return True, None
 
     except Exception as e:
-        process_handler.setup_tracker.remove(pgadmin_process_name)
+        with process_handler.setup_tracker_lock:
+            process_handler.setup_tracker.remove(pgadmin_process_name)
         return False, f"Unhandled exception during pgAdmin setup: {e}"
 
 
@@ -1002,7 +1007,8 @@ def postgres_setup(process_handler=None):
             postgres_config_dir=postgres_config_dir,
             postgres_config_file=postgres_config_file,
         )
-        process_handler.setup_tracker.add(postgres_process_name)
+        with process_handler.setup_tracker_lock:
+            process_handler.setup_tracker.add(postgres_process_name)
         process = process_handler.start_process(
             postgres_process_name, postgres_config_dir, postgres_command
         )