Fix queue review findings Refs #6

importarr/main.py

@@ -32,6 +32,8 @@ app.mount("/static", StaticFiles(directory=str(Path(__file__).parent / "static")
 WORKER_ID = f"importarr-{os.getpid()}"
 _worker_thread: threading.Thread | None = None
 _worker_stop = threading.Event()
+MAX_RETRY_ATTEMPTS = 3
+RETRY_DELAY_SECONDS = 60
 
 
 class ManualBatchCreate(BaseModel):
@@ -260,7 +262,7 @@ def ensure_worker_running() -> None:
     if _worker_thread and _worker_thread.is_alive():
         return
     _worker_stop.clear()
-    state.release_stale_claims(WORKER_ID)
+    state.release_stale_claims()
     _worker_thread = threading.Thread(target=_worker_loop, name="importarr-queue-worker", daemon=True)
     _worker_thread.start()
 
@@ -278,7 +280,6 @@ def _worker_loop() -> None:
             item = state.claim_next_queue_item(WORKER_ID) if queue_accepting_new_jobs() else None
             if item is not None:
                 _import_queue_item(item, importer, from_worker=True)
-                continue
         except Exception:
             state.upsert_queue_item(source_type="system", source_id="queue-worker", name="Queue worker", state="failed", reason="worker loop error")
         _worker_stop.wait(settings.poll_seconds)
@@ -448,16 +449,24 @@ def queue_item_action(item_id: int, payload: QueueItemActionRequest, _: None = D
     if item is None:
         raise HTTPException(status_code=404, detail="queue item not found")
     if payload.action == "retry":
+        if item.get("claimed_by"):
+            raise HTTPException(status_code=409, detail="queue item is currently being imported")
         retry_state = "ready" if item["source_type"] in {"manual", "sab"} else "detected"
         state.mark_queue_item_result(item["source_type"], item["source_id"], retry_state, "retry requested")
     elif payload.action == "run-now":
-        imported = _import_queue_item(item, Importer(settings.movies_root, settings.tv_root), force=True)
+        claimed = state.claim_queue_item(item_id, WORKER_ID, {"ready", "failed", "retrying"})
+        if claimed is None:
+            raise HTTPException(status_code=409, detail="queue item is not available to run now")
+        imported = _import_queue_item(claimed, Importer(settings.movies_root, settings.tv_root), force=True, from_worker=True)
         updated = state.get_queue_item(item_id)
         return {"status": "imported" if imported else "updated", "imported": imported, "item": serialize_queue_item(updated or item)}
     elif payload.action == "ignore":
+        if item.get("claimed_by"):
+            raise HTTPException(status_code=409, detail="queue item is currently being imported")
         state.mark_queue_item_result(item["source_type"], item["source_id"], "skipped", "ignored by user")
     elif payload.action == "remove":
-        state.delete_queue_item(item_id)
+        if not state.delete_queue_item_if_unclaimed(item_id):
+            raise HTTPException(status_code=409, detail="queue item is currently being imported")
         return {"status": "removed", "id": item_id}
     else:
         raise HTTPException(status_code=400, detail="action must be retry, run-now, ignore, or remove")
@@ -526,12 +535,13 @@ def serialize_queue_item(item: dict[str, object]) -> dict[str, object]:
         "completed_at": item["completed_at"],
         "attempt_count": item.get("attempt_count", 0),
         "claimed_by": item.get("claimed_by"),
+        "next_retry_at": item.get("next_retry_at"),
         "sab_status": state_name if source_type == "sab" else None,
         "sab_category": item.get("sab_category") if source_type == "sab" else None,
-        "can_run_now": state_name in {"ready", "manual_batch", "failed", "retrying", "waiting_for_sab"},
+        "can_run_now": state_name in {"ready", "failed", "retrying"} and not item.get("claimed_by"),
         "can_retry": state_name in {"failed", "skipped"},
-        "can_ignore": state_name not in {"imported", "skipped"},
+        "can_ignore": state_name not in {"imported", "skipped", "importing"} and not item.get("claimed_by"),
-        "can_remove": True,
+        "can_remove": not item.get("claimed_by"),
     }
 
 
@@ -657,8 +667,13 @@ def _import_queue_item(item: dict[str, object], importer: Importer, *, force: bo
         return 0
     except Exception as exc:
         state.add_history(source, source, "failed", 0, exc.__class__.__name__)
+        attempts = int(item.get("attempt_count") or 0) + 1
+        if attempts >= MAX_RETRY_ATTEMPTS:
+            state.mark_queue_item_result(str(item["source_type"]), str(item["source_id"]), "failed", exc.__class__.__name__, increment_attempts=True)
+        else:
             next_state = "retrying" if from_worker or force else "failed"
-        state.mark_queue_item_result(str(item["source_type"]), str(item["source_id"]), next_state, exc.__class__.__name__, increment_attempts=True)
+            retry_delay = RETRY_DELAY_SECONDS if next_state == "retrying" else None
+            state.mark_queue_item_result(str(item["source_type"]), str(item["source_id"]), next_state, exc.__class__.__name__, increment_attempts=True, next_retry_seconds=retry_delay)
         return 0
     finally:
         set_current_job(None)
@@ -675,7 +690,7 @@ def _import_manual_batches(importer: Importer) -> int:
             if consume_cancel_request():
                 return imported
             imported += _import_queue_item(item, importer, force=True)
-        if not scan_videos(path):
+        if not scan_videos(path) and not state.batch_has_active_items(batch["id"]):
             state.complete_manual_batch(batch["id"])
     return imported
 
@@ -688,7 +703,7 @@ def startup_queue_worker() -> None:
 @app.on_event("shutdown")
 def shutdown_queue_worker() -> None:
     stop_worker()
-    state.release_stale_claims(WORKER_ID)
+    state.release_stale_claims()
 
 
 def run() -> None:

importarr/state.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import sqlite3
+import threading
 from pathlib import Path
 from typing import Any
 
@@ -20,11 +21,15 @@ class State:
     def __init__(self, path: Path):
         self.path = path
         self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.RLock()
         self.conn = sqlite3.connect(self.path, check_same_thread=False)
         self.conn.row_factory = sqlite3.Row
+        self.conn.execute("pragma journal_mode=WAL")
+        self.conn.execute("pragma busy_timeout = 5000")
         self.migrate()
 
     def migrate(self) -> None:
+        with self._lock:
             self.conn.executescript(
                 """
             create table if not exists manual_batches (
@@ -59,6 +64,7 @@ class State:
               job_id text,
               sab_category text,
               attempt_count integer not null default 0,
+              next_retry_at text,
               last_error text,
               claimed_by text,
               claimed_at text,
@@ -74,6 +80,8 @@ class State:
                 self.conn.execute("alter table import_queue_items add column sab_category text")
             if "attempt_count" not in columns:
                 self.conn.execute("alter table import_queue_items add column attempt_count integer not null default 0")
+            if "next_retry_at" not in columns:
+                self.conn.execute("alter table import_queue_items add column next_retry_at text")
             if "last_error" not in columns:
                 self.conn.execute("alter table import_queue_items add column last_error text")
             if "claimed_by" not in columns:
@@ -83,10 +91,12 @@ class State:
             self.conn.commit()
 
     def get_app_state(self, key: str, default: str | None = None) -> str | None:
+        with self._lock:
             row = self.conn.execute("select value from app_state where key = ?", (key,)).fetchone()
             return row["value"] if row else default
 
     def set_app_state(self, key: str, value: str) -> None:
+        with self._lock:
             self.conn.execute(
                 "insert into app_state(key, value) values (?, ?) on conflict(key) do update set value=excluded.value",
                 (key, value),
@@ -94,11 +104,13 @@ class State:
             self.conn.commit()
 
     def add_manual_batch(self, path: Path) -> dict[str, Any]:
+        with self._lock:
             self.conn.execute("insert or ignore into manual_batches(path) values (?)", (str(path),))
             self.conn.commit()
         return self.get_manual_batch_by_path(path)
 
     def get_manual_batch_by_path(self, path: Path) -> dict[str, Any]:
+        with self._lock:
             row = self.conn.execute("select * from manual_batches where path = ?", (str(path),)).fetchone()
             return dict(row)
 
@@ -107,18 +119,22 @@ class State:
         if active_only:
             sql += " where status = 'active'"
         sql += " order by created_at desc"
+        with self._lock:
             return [dict(row) for row in self.conn.execute(sql)]
 
     def delete_manual_batch(self, batch_id: int) -> None:
+        with self._lock:
             self.conn.execute("delete from manual_batches where id = ?", (batch_id,))
             self.conn.execute("delete from import_queue_items where batch_id = ? and source_type = 'manual'", (batch_id,))
             self.conn.commit()
 
     def complete_manual_batch(self, batch_id: int) -> None:
+        with self._lock:
             self.conn.execute("update manual_batches set status='completed', completed_at=current_timestamp where id=?", (batch_id,))
             self.conn.commit()
 
     def add_history(self, source: Path, target: Path, status: str, bytes_count: int = 0, error: str | None = None) -> None:
+        with self._lock:
             self.conn.execute(
                 "insert into import_history(source,target,status,bytes,error,completed_at) values (?,?,?,?,?,case when ? in ('imported','failed') then current_timestamp else null end)",
                 (str(source), str(target), status, bytes_count, error, status),
@@ -141,6 +157,7 @@ class State:
         sab_category: str | None = None,
         preserve_finished_state: bool = True,
     ) -> dict[str, Any]:
+        with self._lock:
             self.conn.execute(
                 """
             insert into import_queue_items(source_type, source_id, source_path, name, state, reason, relative_path, size, batch_id, job_id, sab_category)
@@ -162,6 +179,11 @@ class State:
               job_id=excluded.job_id,
               sab_category=excluded.sab_category,
               updated_at=current_timestamp,
+              next_retry_at=case
+                when excluded.state = 'retrying' then coalesce(import_queue_items.next_retry_at, excluded.next_retry_at)
+                when excluded.state = 'ready' then null
+                else import_queue_items.next_retry_at
+              end,
               completed_at=case
                 when ? and import_queue_items.state in ('imported','failed','skipped') then import_queue_items.completed_at
                 when excluded.state in ('imported','failed','skipped') then current_timestamp
@@ -175,18 +197,23 @@ class State:
             return dict(row)
 
     def mark_queue_item(self, source_type: str, source_id: str, state: str, reason: str | None = None) -> None:
+        with self._lock:
             self.conn.execute(
-            "update import_queue_items set state=?, reason=?, updated_at=current_timestamp, claimed_by=null, claimed_at=null, completed_at=case when ? in ('imported','failed','skipped') then current_timestamp else completed_at end where source_type=? and source_id=?",
+                "update import_queue_items set state=?, reason=?, updated_at=current_timestamp, claimed_by=null, claimed_at=null, next_retry_at=case when ?='ready' then null else next_retry_at end, completed_at=case when ? in ('imported','failed','skipped') then current_timestamp else completed_at end where source_type=? and source_id=?",
-            (state, reason, state, source_type, source_id),
+                (state, reason, state, state, source_type, source_id),
             )
             self.conn.commit()
 
     def claim_next_queue_item(self, worker_id: str) -> dict[str, Any] | None:
-        with self.conn:
+        with self._lock:
             row = self.conn.execute(
                 """
                 select * from import_queue_items
-                where state in ('ready','retrying') and claimed_by is null
+                where claimed_by is null
+                  and (
+                    state = 'ready'
+                    or (state = 'retrying' and (next_retry_at is null or next_retry_at <= current_timestamp))
+                  )
                 order by
                   case state when 'ready' then 0 else 1 end,
                   updated_at asc,
@@ -207,19 +234,28 @@ class State:
             claimed = self.conn.execute("select * from import_queue_items where id = ?", (row["id"],)).fetchone()
             return dict(claimed) if claimed and claimed["claimed_by"] == worker_id else None
 
-    def release_stale_claims(self, worker_id: str | None = None) -> int:
+    def release_stale_claims(self) -> int:
-        if worker_id is None:
+        with self._lock:
             cursor = self.conn.execute(
                 "update import_queue_items set state='retrying', claimed_by=null, claimed_at=null, updated_at=current_timestamp where state='importing'"
             )
-        else:
-            cursor = self.conn.execute(
-                "update import_queue_items set state='retrying', claimed_by=null, claimed_at=null, updated_at=current_timestamp where state='importing' and claimed_by=?",
-                (worker_id,),
-            )
             self.conn.commit()
             return cursor.rowcount
 
+    def claim_queue_item(self, item_id: int, worker_id: str, allowed_states: set[str]) -> dict[str, Any] | None:
+        placeholders = ",".join("?" for _ in allowed_states)
+        with self._lock:
+            row = self.conn.execute(f"select * from import_queue_items where id = ? and state in ({placeholders}) and claimed_by is null", (item_id, *allowed_states)).fetchone()
+            if row is None:
+                return None
+            self.conn.execute(
+                "update import_queue_items set state='importing', claimed_by=?, claimed_at=current_timestamp, updated_at=current_timestamp where id=? and claimed_by is null",
+                (worker_id, item_id),
+            )
+            self.conn.commit()
+            claimed = self.conn.execute("select * from import_queue_items where id = ?", (item_id,)).fetchone()
+            return dict(claimed) if claimed and claimed["claimed_by"] == worker_id else None
+
     def mark_queue_item_result(
         self,
         source_type: str,
@@ -228,7 +264,9 @@ class State:
         reason: str | None = None,
         *,
         increment_attempts: bool = False,
+        next_retry_seconds: int | None = None,
     ) -> None:
+        with self._lock:
             self.conn.execute(
                 """
             update import_queue_items
@@ -236,22 +274,35 @@ class State:
                 reason=?,
                 last_error=case when ? in ('failed','retrying','skipped') then ? else null end,
                 attempt_count=attempt_count + ?,
+                next_retry_at=case
+                  when ? = 'retrying' and ? is not null then datetime('now', '+' || ? || ' seconds')
+                  when ? in ('ready','imported','failed','skipped') then null
+                  else next_retry_at
+                end,
                 claimed_by=null,
                 claimed_at=null,
                 updated_at=current_timestamp,
                 completed_at=case when ? in ('imported','failed','skipped') then current_timestamp else null end
             where source_type=? and source_id=?
             """,
-            (state, reason, state, reason, 1 if increment_attempts else 0, state, source_type, source_id),
+                (state, reason, state, reason, 1 if increment_attempts else 0, state, next_retry_seconds, next_retry_seconds, state, state, source_type, source_id),
             )
             self.conn.commit()
 
     def delete_queue_item(self, item_id: int) -> bool:
+        with self._lock:
             cursor = self.conn.execute("delete from import_queue_items where id = ?", (item_id,))
             self.conn.commit()
             return cursor.rowcount > 0
 
+    def delete_queue_item_if_unclaimed(self, item_id: int) -> bool:
+        with self._lock:
+            cursor = self.conn.execute("delete from import_queue_items where id = ? and claimed_by is null", (item_id,))
+            self.conn.commit()
+            return cursor.rowcount > 0
+
     def delete_queue_items_by_state(self, source_type: str, state: str, reason: str | None = None) -> int:
+        with self._lock:
             if reason is None:
                 cursor = self.conn.execute("delete from import_queue_items where source_type = ? and state = ?", (source_type, state))
             else:
@@ -260,22 +311,34 @@ class State:
             return cursor.rowcount
 
     def get_queue_item(self, item_id: int) -> dict[str, Any] | None:
+        with self._lock:
             row = self.conn.execute("select * from import_queue_items where id = ?", (item_id,)).fetchone()
             return dict(row) if row else None
 
     def remove_missing_manual_items(self, batch_id: int, source_ids: set[str]) -> None:
-        rows = self.conn.execute("select source_id from import_queue_items where source_type='manual' and batch_id=?", (batch_id,)).fetchall()
+        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()
             for row in rows:
-            if row["source_id"] not in source_ids:
+                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"],))
             self.conn.commit()
 
+    def batch_has_active_items(self, batch_id: int) -> bool:
+        with self._lock:
+            row = self.conn.execute(
+                "select 1 from import_queue_items where batch_id = ? and source_type='manual' and state not in ('imported','failed','skipped') limit 1",
+                (batch_id,),
+            ).fetchone()
+            return row is not None
+
     def list_queue_items(self, active_only: bool = True) -> list[dict[str, Any]]:
         sql = "select * from import_queue_items"
         if active_only:
             sql += " where state not in ('imported','failed','skipped')"
         sql += " order by updated_at desc, id desc"
+        with self._lock:
             return [dict(row) for row in self.conn.execute(sql)]
 
     def list_history(self, limit: int = 100) -> list[dict[str, Any]]:
+        with self._lock:
             return [dict(row) for row in self.conn.execute("select * from import_history order by id desc limit ?", (limit,))]

tests/test_queue_controls.py

@@ -230,6 +230,90 @@ def test_worker_claimed_failure_retries_item(tmp_path, monkeypatch):
     assert imported == 0
     assert updated["state"] == "retrying"
     assert updated["attempt_count"] == 1
+    assert updated["next_retry_at"] is not None
+
+
+def test_startup_releases_stale_claims_from_previous_worker(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")
+    main.state.claim_queue_item(row["id"], "old-worker", {"ready"})
+
+    main.ensure_worker_running()
+    updated = main.state.get_queue_item(row["id"])
+    main.stop_worker()
+
+    assert updated["state"] == "retrying"
+    assert updated["claimed_by"] is None
+
+
+def test_run_now_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")
+    main.state.claim_queue_item(row["id"], "other-worker", {"ready"})
+
+    try:
+        main.queue_item_action(row["id"], main.QueueItemActionRequest(action="run-now"))
+    except main.HTTPException as exc:
+        assert exc.status_code == 409
+    else:
+        raise AssertionError("expected HTTPException")
+
+
+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")
+    main.state.claim_queue_item(row["id"], "other-worker", {"ready"})
+
+    try:
+        main.queue_item_action(row["id"], main.QueueItemActionRequest(action="remove"))
+    except main.HTTPException as exc:
+        assert exc.status_code == 409
+    else:
+        raise AssertionError("expected HTTPException")
+
+
+def test_manual_import_completion_persists_completed_row_and_batch_completion(tmp_path, monkeypatch):
+    main, download, movies, tv = configure_main(tmp_path, monkeypatch)
+    batch = download / "Release"
+    batch.mkdir(parents=True)
+    source = batch / "Done.mkv"
+    source.write_bytes(b"done")
+    created_batch = main.state.add_manual_batch(batch)
+
+    imported = main._import_manual_batches(main.Importer(movies, tv))
+    main.sync_manual_queue()
+
+    assert imported == 1
+    rows = {row["name"]: row for row in main.state.list_queue_items(active_only=False)}
+    assert rows["Done.mkv"]["state"] == "imported"
+    assert main.state.get_queue_item(rows["Done.mkv"]["id"])["state"] == "imported"
+    batch_row = next(batch for batch in main.state.list_manual_batches() if batch["id"] == created_batch["id"])
+    assert batch_row["status"] == "completed"
+
+
+def test_worker_failure_stops_retrying_after_limit(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")
+    row = main.state.upsert_queue_item(source_type="manual", source_id=str(source), source_path=source, name=source.name, state="ready")
+
+    class BrokenImporter:
+        def import_file(self, *args, **kwargs):
+            raise RuntimeError("boom")
+
+    for _ in range(main.MAX_RETRY_ATTEMPTS):
+        claimed = main.state.claim_queue_item(row["id"], main.WORKER_ID, {"ready", "retrying"})
+        assert claimed is not None
+        main._import_queue_item(claimed, BrokenImporter(), from_worker=True)
+        current = main.state.get_queue_item(row["id"])
+        if current["state"] == "retrying":
+            main.state.mark_queue_item("manual", current["source_id"], "ready", "retry window elapsed")
+
+    updated = main.state.get_queue_item(row["id"])
+    assert updated["state"] == "failed"
+    assert updated["attempt_count"] == main.MAX_RETRY_ATTEMPTS
 
 
 def test_control_update_runs_configured_command(tmp_path, monkeypatch):

tests/test_state.py

@@ -39,9 +39,41 @@ def test_release_stale_claims_requeues_importing_items(tmp_path):
     state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")
     state.claim_next_queue_item("worker-1")
 
-    released = state.release_stale_claims("worker-1")
+    released = state.release_stale_claims()
     row = state.get_queue_item(1)
 
     assert released == 1
     assert row["state"] == "retrying"
     assert row["claimed_by"] is None
+
+
+def test_claim_queue_item_requires_unclaimed_allowed_state(tmp_path):
+    state = State(tmp_path / "state.db")
+    row = state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")
+
+    claimed = state.claim_queue_item(row["id"], "worker-1", {"ready"})
+    blocked = state.claim_queue_item(row["id"], "worker-2", {"ready", "importing"})
+
+    assert claimed is not None
+    assert claimed["claimed_by"] == "worker-1"
+    assert blocked is None
+
+
+def test_remove_missing_manual_items_keeps_terminal_rows(tmp_path):
+    state = State(tmp_path / "state.db")
+    state.upsert_queue_item(source_type="manual", source_id="done", name="Done.mkv", state="imported", batch_id=1)
+    state.upsert_queue_item(source_type="manual", source_id="pending", name="Pending.mkv", state="ready", batch_id=1)
+
+    state.remove_missing_manual_items(1, set())
+
+    rows = {row["source_id"]: row for row in state.list_queue_items(active_only=False)}
+    assert "done" in rows
+    assert "pending" not in rows
+
+
+def test_retry_item_respects_next_retry_at(tmp_path):
+    state = State(tmp_path / "state.db")
+    state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")
+    state.mark_queue_item_result("manual", "a", "retrying", "boom", increment_attempts=True, next_retry_seconds=60)
+
+    assert state.claim_next_queue_item("worker-1") is None