fix(musicseerr): implement persistent job database and startup recovery to survive pod restarts

apps/musicseerr/main.py

@@ -15,8 +15,30 @@ logger = logging.getLogger("musicseerr-main")
 
 app = FastAPI(title="MusicSeerr Ingestion Pipeline", version="1.0.0")
 
-# In-memory jobs database
+import json
-JOBS = {}
+
+JOBS_FILE = "/remote-music/.musicseerr_jobs.json"
+
+def save_jobs_db(jobs_dict):
+    try:
+        temp_file = JOBS_FILE + ".tmp"
+        with open(temp_file, "w") as f:
+            json.dump(jobs_dict, f, indent=2)
+        os.replace(temp_file, JOBS_FILE)
+    except Exception as e:
+        logger.error(f"Failed to save jobs database: {e}")
+
+def load_jobs_db():
+    if os.path.exists(JOBS_FILE):
+        try:
+            with open(JOBS_FILE, "r") as f:
+                return json.load(f)
+        except Exception as e:
+            logger.error(f"Failed to load jobs database: {e}")
+    return {}
+
+# Load jobs database
+JOBS = load_jobs_db()
 
 # Read configurations from Environment Variables
 PROXY_URL = os.getenv("GLUETUN_PROXY_URL", "http://gluetun-svc:8888")
@@ -79,8 +101,10 @@ def submit_job(req: SubmitRequest, background_tasks: BackgroundTasks):
         "status": "pending",
         "priority_used": None,
         "files": [],
-        "error": None
+        "error": None,
+        "chat_id": None
     }
+    save_jobs_db(JOBS)
     
     background_tasks.add_task(run_download_task, task_id, query, PROXY_URL, JOBS)
     
@@ -123,8 +147,10 @@ async def telegram_webhook(request: Request, background_tasks: BackgroundTasks):
             "status": "pending",
             "priority_used": None,
             "files": [],
-            "error": None
+            "error": None,
+            "chat_id": chat_id
         }
+        save_jobs_db(JOBS)
         
         # Pass chat_id to the background task (we will update run_download_task to support this)
         background_tasks.add_task(run_download_task, task_id, text, PROXY_URL, JOBS, chat_id)
@@ -546,8 +572,10 @@ def run_lastfm_auto_download(username: str):
                 "status": "pending",
                 "priority_used": None,
                 "files": [],
-                "error": None
+                "error": None,
+                "chat_id": None
             }
+            save_jobs_db(JOBS)
             try:
                 run_download_task(task_id, track_query, PROXY_URL, JOBS)
             except Exception as e:
@@ -582,6 +610,45 @@ async def lastfm_scheduler():
         await asyncio.sleep(interval_hours * 3600)
 
 
+async def recover_jobs():
+    # Wait for remote music directory mount to be fully ready
+    logger.info("Starting recovery checks for interrupted/pending ingestion tasks...")
+    await asyncio.sleep(10)
+    
+    global JOBS
+    JOBS = load_jobs_db()
+    
+    recovered_count = 0
+    for task_id, job in list(JOBS.items()):
+        if job.get("status") in ("pending", "downloading"):
+            logger.info(f"Recovering interrupted job {task_id}: '{job.get('query')}'")
+            job["status"] = "pending"
+            job["error"] = "Interrupted by server restart, resuming..."
+            
+            # Re-queue the task in a background thread executor using the event loop
+            try:
+                loop = asyncio.get_event_loop()
+                loop.run_in_executor(
+                    None, 
+                    run_download_task, 
+                    task_id, 
+                    job.get("query"), 
+                    PROXY_URL, 
+                    JOBS, 
+                    job.get("chat_id")
+                )
+                recovered_count += 1
+            except Exception as e:
+                logger.error(f"Failed to recover job {task_id}: {e}")
+                
+    if recovered_count > 0:
+        save_jobs_db(JOBS)
+        logger.info(f"Successfully recovered and resumed {recovered_count} tasks.")
+    else:
+        logger.info("No pending or interrupted ingestion tasks found to recover.")
+
+
 @app.on_event("startup")
 async def startup_event():
     asyncio.create_task(lastfm_scheduler())
+    asyncio.create_task(recover_jobs())

apps/musicseerr/tasks.py

@@ -18,6 +18,17 @@ DOWNLOAD_SEMAPHORE = threading.Semaphore(5)
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger("musicseerr-tasks")
 
+JOBS_FILE = "/remote-music/.musicseerr_jobs.json"
+
+def save_jobs(jobs_dict):
+    try:
+        temp_file = JOBS_FILE + ".tmp"
+        with open(temp_file, "w") as f:
+            json.dump(jobs_dict, f, indent=2)
+        os.replace(temp_file, JOBS_FILE)
+    except Exception as e:
+        logger.error(f"Failed to save jobs database: {e}")
+
 def move_new_files_to_dest(temp_dir: str, dest_dir: str) -> list:
     moved = []
     if not os.path.exists(temp_dir):
@@ -495,6 +506,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict:
     Core download manager task that runs in the background.
     """
     jobs_dict[task_id]["status"] = "downloading"
+    save_jobs(jobs_dict)
     
     temp_dir = f"/tmp/downloads/{task_id}"
     os.makedirs(temp_dir, exist_ok=True)
@@ -539,6 +551,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict:
 
         if is_playlist:
             jobs_dict[task_id]["priority_used"] = "playlist-scraper"
+            save_jobs(jobs_dict)
             logger.info(f"[{task_id}] Spotify playlist detected: '{query}'")
             track_queries = scrape_spotify_playlist(query, resolved_proxy_url)
             if not track_queries:
@@ -619,6 +632,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict:
                     if "files" not in jobs_dict[task_id] or not jobs_dict[task_id]["files"]:
                         jobs_dict[task_id]["files"] = []
                     jobs_dict[task_id]["files"].extend(new_moved)
+                    save_jobs(jobs_dict)
                     logger.info(f"[{task_id}] Successfully downloaded and moved {len(new_moved)} files for track '{track_query}'")
                     
             if download_count > 0:
@@ -635,6 +649,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict:
             # Priority 1: spotDL (text search or Spotify URL)
             if not is_tubidy and not is_youtube:
                 jobs_dict[task_id]["priority_used"] = "spotDL"
+                save_jobs(jobs_dict)
                 logger.info(f"[{task_id}] Attempting Priority 1: spotDL for '{query}'")
                 
                 env = os.environ.copy()
@@ -660,6 +675,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict:
             # Priority 2: yt-dlp (direct YouTube or fallback for failed spotDL)
             if not is_tubidy and not is_spotify:
                 jobs_dict[task_id]["priority_used"] = "yt-dlp"
+                save_jobs(jobs_dict)
                 logger.info(f"[{task_id}] Attempting Priority 2: yt-dlp for '{query}'")
                 
                 env = os.environ.copy()
@@ -697,6 +713,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict:
             # Priority 3: Custom Tubidy Link Scraper (if URL is directly passed)
             if is_tubidy:
                 jobs_dict[task_id]["priority_used"] = "Tubidy"
+                save_jobs(jobs_dict)
                 logger.info(f"[{task_id}] Attempting Priority 3: Tubidy Scraper for '{query}'")
                 
                 # Scrape MP3 URL
@@ -725,6 +742,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict:
             # Priority 4: Tubidy Search Fallback (if spotDL and yt-dlp failed and no files are present)
             if not download_successful and not is_tubidy:
                 jobs_dict[task_id]["priority_used"] = "Tubidy-Search"
+                save_jobs(jobs_dict)
                 logger.info(f"[{task_id}] Attempting Fallback Priority 4: Tubidy Search Scraper for '{query}'")
                 
                 search_query = query
@@ -780,11 +798,13 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict:
             logger.info(f"[{task_id}] All tracks in this query were already downloaded (skipped duplicates).")
             jobs_dict[task_id]["files"] = []
             jobs_dict[task_id]["status"] = "completed"
+            save_jobs(jobs_dict)
             if chat_id:
                 send_telegram_notification(chat_id, f"✅ Successfully ingested request: '{query}'\n\n(All tracks already existed - skipped duplicates)")
             return
             
         jobs_dict[task_id]["status"] = "completed"
+        save_jobs(jobs_dict)
         logger.info(f"[{task_id}] Music Ingestion task fully completed.")
         
         if chat_id:
@@ -795,6 +815,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict:
         logger.error(f"[{task_id}] Ingestion failed: {e}")
         jobs_dict[task_id]["status"] = "failed"
         jobs_dict[task_id]["error"] = str(e)
+        save_jobs(jobs_dict)
         if chat_id:
             send_telegram_notification(chat_id, f"❌ Failed to ingest request: '{query}'\nError: {e}")