From c5cc0901c5818d23a1d081606390fe80f5fddb48 Mon Sep 17 00:00:00 2001 From: Daniel Gradman-Svendsen Date: Thu, 30 Jul 2026 10:40:39 +0200 Subject: [PATCH] Fix remaining queue races Refs #6 --- importarr/main.py | 7 +++-- importarr/state.py | 6 ++-- tests/test_queue_controls.py | 58 +++++++++++++++++++++++++++++++----- 3 files changed, 59 insertions(+), 12 deletions(-) diff --git a/importarr/main.py b/importarr/main.py index 6a8f684..1d2fe23 100644 --- a/importarr/main.py +++ b/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) diff --git a/importarr/state.py b/importarr/state.py index d191f04..2d688e5 100644 --- a/importarr/state.py +++ b/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: diff --git a/tests/test_queue_controls.py b/tests/test_queue_controls.py index 6ce7dce..62fd390 100644 --- a/tests/test_queue_controls.py +++ b/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")