Fix remaining queue races Refs #6

importarr/main.py

@@ -30,6 +30,7 @@ templates = Jinja2Templates(directory=str(Path(__file__).parent / "templates"))
 app = FastAPI(title="Importarr")
 app.mount("/static", StaticFiles(directory=str(Path(__file__).parent / "static")), name="static")
 WORKER_ID = f"importarr-{os.getpid()}"
+WORKER_SHUTDOWN_TIMEOUT_SECONDS = 5.0
 _worker_thread: threading.Thread | None = None
 _worker_stop = threading.Event()
 MAX_RETRY_ATTEMPTS = 3
@@ -271,7 +272,7 @@ def stop_worker(*, wait: bool = False) -> bool:
     _worker_stop.set()
     thread = _worker_thread
     if wait and thread and thread.is_alive() and thread is not threading.current_thread():
-        thread.join()
+        thread.join(WORKER_SHUTDOWN_TIMEOUT_SECONDS)
     return not thread or not thread.is_alive()
 
 
@@ -622,10 +623,12 @@ async def _import_ready_sab_jobs(importer: Importer, force: bool = False) -> int
         readiness = classify_history_item(item, active, settings.sab_category, settings.download_root, force_status=force, sab_storage_root=settings.sab_storage_root)
         if readiness.storage is None or (not readiness.ready and not force):
             continue
+        job_id = str(item.get("nzo_id") or item.get("nzoid") or item.get("name") or "")
+        sab_category = str(item.get("category") or item.get("cat") or "")
         for video in scan_videos(readiness.storage):
             if consume_cancel_request():
                 return imported
-            row = state.upsert_queue_item(source_type="sab", source_id=str(video.path), source_path=video.path, name=video.path.name, state="ready", reason=readiness.reason, relative_path=str(video.relative_path), size=video.size)
+            row = state.upsert_queue_item(source_type="sab", source_id=str(video.path), source_path=video.path, name=video.path.name, state="ready", reason=readiness.reason, relative_path=str(video.relative_path), size=video.size, job_id=job_id, sab_category=sab_category)
             claimed = state.claim_queue_item(row["id"], WORKER_ID, {"ready", "failed", "retrying"})
             if claimed is not None:
                 imported += _import_queue_item(claimed, importer, force=force, from_worker=True)

importarr/state.py

@@ -313,10 +313,10 @@ class State:
 
     def remove_missing_manual_items(self, batch_id: int, source_ids: set[str]) -> None:
         with self._lock:
-            rows = self.conn.execute("select source_id, state from import_queue_items where source_type='manual' and batch_id=?", (batch_id,)).fetchall()
+            rows = self.conn.execute("select source_id, state, claimed_by from import_queue_items where source_type='manual' and batch_id=?", (batch_id,)).fetchall()
             for row in rows:
-                if row["source_id"] not in source_ids and row["state"] not in TERMINAL_QUEUE_STATES:
-                    self.conn.execute("delete from import_queue_items where source_type='manual' and source_id=?", (row["source_id"],))
+                if row["source_id"] not in source_ids and row["state"] not in TERMINAL_QUEUE_STATES and not row["claimed_by"]:
+                    self.conn.execute("delete from import_queue_items where source_type='manual' and source_id=? and claimed_by is null", (row["source_id"],))
             self.conn.commit()
 
     def batch_has_active_items(self, batch_id: int) -> bool:

tests/test_queue_controls.py

@@ -1,3 +1,4 @@
+import asyncio
 import threading
 
 from importarr.config import Settings
@@ -316,7 +317,7 @@ def test_ignore_loses_atomic_race_with_worker_claim(tmp_path, monkeypatch):
     assert main.state.get_queue_item(row["id"])["claimed_by"] == "worker"
 
 
-def test_shutdown_waits_for_active_import_before_releasing_claims(tmp_path, monkeypatch):
+def test_shutdown_timeout_does_not_release_live_worker_claim(tmp_path, monkeypatch):
     main, download, _movies, _tv = configure_main(tmp_path, monkeypatch)
     source = download / "A.mkv"
     source.parent.mkdir(parents=True)
@@ -337,21 +338,64 @@ def test_shutdown_waits_for_active_import_before_releasing_claims(tmp_path, monk
 
     worker = threading.Thread(target=active_worker)
     monkeypatch.setattr(main, "_worker_thread", worker)
+    monkeypatch.setattr(main, "WORKER_SHUTDOWN_TIMEOUT_SECONDS", 0.01)
     worker.start()
     assert started.wait(1)
-    shutdown = threading.Thread(target=main.shutdown_queue_worker)
-    shutdown.start()
 
-    shutdown.join(0.05)
-    assert shutdown.is_alive()
+    main.shutdown_queue_worker()
+
+    assert worker.is_alive()
     assert main.state.get_queue_item(row["id"])["claimed_by"] == main.WORKER_ID
 
     finish.set()
-    shutdown.join(1)
-    assert not shutdown.is_alive()
+    worker.join(1)
+    assert not worker.is_alive()
     assert main.state.get_queue_item(row["id"])["claimed_by"] is None
 
 
+def test_manual_sync_preserves_claim_after_source_is_unlinked(tmp_path, monkeypatch):
+    main, download, _movies, _tv = configure_main(tmp_path, monkeypatch)
+    batch = download / "Release"
+    batch.mkdir(parents=True)
+    source = batch / "A.mkv"
+    source.write_bytes(b"a")
+    created_batch = main.state.add_manual_batch(batch)
+    row = main.state.upsert_queue_item(source_type="manual", source_id=str(source), source_path=source, name=source.name, state="ready", batch_id=created_batch["id"])
+    main.state.claim_queue_item(row["id"], main.WORKER_ID, {"ready"})
+
+    source.unlink()
+    main.sync_manual_queue()
+
+    preserved = main.state.get_queue_item(row["id"])
+    assert preserved["state"] == "importing"
+    assert preserved["claimed_by"] == main.WORKER_ID
+
+
+def test_bulk_sab_import_preserves_job_metadata(tmp_path, monkeypatch):
+    main, download, movies, tv = configure_main(tmp_path, monkeypatch)
+    release = download / "Release"
+    release.mkdir(parents=True)
+    source = release / "Movie.mkv"
+    source.write_bytes(b"movie")
+
+    class FakeSabnzbdClient:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        async def active_nzo_ids(self):
+            return set()
+
+        async def history(self):
+            return {"history": {"slots": [{"nzo_id": "SAB-123", "name": "Release", "category": "manual", "status": "Completed", "storage": str(release)}]}}
+
+    monkeypatch.setattr(main, "SabnzbdClient", FakeSabnzbdClient)
+
+    assert asyncio.run(main._import_ready_sab_jobs(main.Importer(movies, tv))) == 1
+    row = main.state.list_queue_items(active_only=False)[0]
+    assert row["job_id"] == "SAB-123"
+    assert row["sab_category"] == "manual"
+
+
 def test_remove_conflicts_when_item_is_claimed(tmp_path, monkeypatch):
     main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
     row = main.state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")