Implement persistent queue engine Refs #6

importarr/main.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import json
+import os
 from pathlib import Path
 import subprocess
+import threading
 import time
 from typing import Annotated
 
@@ -27,6 +29,9 @@ state = State(settings.state_path)
 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_thread: threading.Thread | None = None
+_worker_stop = threading.Event()
 
 
 class ManualBatchCreate(BaseModel):
@@ -107,6 +112,10 @@ def index(request: Request) -> HTMLResponse:
 def status() -> dict[str, object]:
     history = state.list_history()
     control = control_status()
+    queue_items = state.list_queue_items(active_only=False)
+    queue_counts: dict[str, int] = {}
+    for row in queue_items:
+        queue_counts[row["state"]] = queue_counts.get(row["state"], 0) + 1
     return {
         "app": "Importarr",
         "build": build_info(),
@@ -125,6 +134,8 @@ def status() -> dict[str, object]:
         "manual_batches": len(state.list_manual_batches(active_only=True)),
         "imported_total": sum(1 for row in history if row["status"] == "imported"),
         "failed_total": sum(1 for row in history if row["status"] == "failed"),
+        "queue_total": len(queue_items),
+        "queue_counts": queue_counts,
         "current": control["current"],
         "control": control,
     }
@@ -239,11 +250,40 @@ def set_current_job(name: str | None, *, bytes_copied: int = 0, total_bytes: int
     percent = round((bytes_copied / total_bytes * 100), 2) if total_bytes else 0
     state.set_app_state(
         "current_job",
-        json.dumps({"file": name, "bytes_copied": bytes_copied, "total_bytes": total_bytes, "percent": percent, "started_at": started}),
+        json.dumps({"file": name, "name": Path(name).name if name else name, "bytes_copied": bytes_copied, "total_bytes": total_bytes, "percent": percent, "started_at": started}),
     )
     return started
 
 
+def ensure_worker_running() -> None:
+    global _worker_thread
+    if _worker_thread and _worker_thread.is_alive():
+        return
+    _worker_stop.clear()
+    state.release_stale_claims(WORKER_ID)
+    _worker_thread = threading.Thread(target=_worker_loop, name="importarr-queue-worker", daemon=True)
+    _worker_thread.start()
+
+
+def stop_worker() -> None:
+    _worker_stop.set()
+
+
+def _worker_loop() -> None:
+    importer = Importer(settings.movies_root, settings.tv_root)
+    while not _worker_stop.is_set():
+        try:
+            if queue_accepting_new_jobs():
+                sync_manual_queue()
+            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)
+
+
 @app.post("/api/control/queue")
 def set_queue_control(payload: QueueControlRequest, _: None = Depends(require_write_auth)) -> dict[str, object]:
     if payload.mode not in {"running", "paused", "stopped"}:
@@ -251,6 +291,9 @@ def set_queue_control(payload: QueueControlRequest, _: None = Depends(require_wr
     state.set_app_state("queue_mode", payload.mode)
     if payload.mode == "running":
         state.set_app_state("cancel_requested", "false")
+        ensure_worker_running()
+    elif payload.mode == "stopped":
+        stop_worker()
     return control_status()
 
 
@@ -258,6 +301,7 @@ def set_queue_control(payload: QueueControlRequest, _: None = Depends(require_wr
 def start_queue(_: None = Depends(require_write_auth)) -> dict[str, object]:
     state.set_app_state("queue_mode", "running")
     state.set_app_state("cancel_requested", "false")
+    ensure_worker_running()
     return {"control": control_status(), "command_result": _run_control_command(settings.start_command)}
 
 
@@ -271,6 +315,7 @@ def pause_queue(_: None = Depends(require_write_auth)) -> dict[str, object]:
 def stop_queue(_: None = Depends(require_write_auth)) -> dict[str, object]:
     state.set_app_state("queue_mode", "stopped")
     state.set_app_state("cancel_requested", "true")
+    stop_worker()
     return {"control": control_status(), "command_result": _run_control_command(settings.stop_command)}
 
 
@@ -403,19 +448,19 @@ 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":
-        retry_state = "manual_batch" if item["source_type"] == "manual" else "ready"
+        retry_state = "ready" if item["source_type"] in {"manual", "sab"} else "detected"
-        state.mark_queue_item(item["source_type"], item["source_id"], retry_state, "retry requested")
+        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))
+        imported = _import_queue_item(item, Importer(settings.movies_root, settings.tv_root), force=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":
-        state.mark_queue_item(item["source_type"], item["source_id"], "skipped", "ignored by user")
+        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)
         return {"status": "removed", "id": item_id}
     else:
-        raise HTTPException(status_code=400, detail="action must be retry, ignore, or remove")
+        raise HTTPException(status_code=400, detail="action must be retry, run-now, ignore, or remove")
     updated = state.get_queue_item(item_id)
     return {"status": "updated", "item": serialize_queue_item(updated or item)}
 
@@ -443,7 +488,6 @@ async def sync_queue() -> None:
         state.upsert_queue_item(source_type="system", source_id="sab-sync", name="SABnzbd", state="failed", reason=exc.__class__.__name__)
         return
     slots = data.get("history", {}).get("slots", [])
-    state.delete_queue_items_by_state("sab", "ignored")
     for item in slots:
         readiness = classify_history_item(item, active, settings.sab_category, settings.download_root, sab_storage_root=settings.sab_storage_root)
         job_id = str(item.get("nzo_id") or item.get("nzoid") or item.get("name") or "")
@@ -453,7 +497,8 @@ async def sync_queue() -> None:
             for video in scan_videos(readiness.storage):
                 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=str(item.get("category") or item.get("cat") or ""))
         else:
-            state.upsert_queue_item(source_type="sab", source_id=job_id, source_path=readiness.storage, name=str(item.get("name") or job_id), state=readiness.state, reason=readiness.reason, job_id=job_id, sab_category=str(item.get("category") or item.get("cat") or ""))
+            pending_state = "waiting_for_sab" if readiness.state not in {"failed", "skipped"} else readiness.state
+            state.upsert_queue_item(source_type="sab", source_id=job_id, source_path=readiness.storage, name=str(item.get("name") or job_id), state=pending_state, reason=readiness.reason, job_id=job_id, sab_category=str(item.get("category") or item.get("cat") or ""))
 
 
 def queue_jobs() -> list[dict[str, object]]:
@@ -479,9 +524,11 @@ def serialize_queue_item(item: dict[str, object]) -> dict[str, object]:
         "first_seen_at": item["first_seen_at"],
         "updated_at": item["updated_at"],
         "completed_at": item["completed_at"],
+        "attempt_count": item.get("attempt_count", 0),
+        "claimed_by": item.get("claimed_by"),
         "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"},
+        "can_run_now": state_name in {"ready", "manual_batch", "failed", "retrying", "waiting_for_sab"},
         "can_retry": state_name in {"failed", "skipped"},
         "can_ignore": state_name not in {"imported", "skipped"},
         "can_remove": True,
@@ -491,8 +538,12 @@ def serialize_queue_item(item: dict[str, object]) -> dict[str, object]:
 def job_group(state_name: str, source_type: str) -> str:
     if source_type == "manual":
         return "manual_batch"
+    if state_name in {"detected", "waiting_for_sab"}:
+        return "sab_processing"
     if state_name == "ready":
         return "ready"
+    if state_name == "retrying":
+        return "failed"
     if state_name in {"importing", "copying"}:
         return "importing"
     if state_name == "failed":
@@ -531,7 +582,7 @@ def sync_manual_queue() -> None:
         for video in scan_videos(Path(batch["path"])):
             source_id = str(video.path)
             seen.add(source_id)
-            state.upsert_queue_item(source_type="manual", source_id=source_id, source_path=video.path, name=video.path.name, state="manual_batch", relative_path=str(video.path.relative_to(root)), size=video.size, batch_id=batch["id"])
+            state.upsert_queue_item(source_type="manual", source_id=source_id, source_path=video.path, name=video.path.name, state="ready", reason="manual batch detected", relative_path=str(video.path.relative_to(root)), size=video.size, batch_id=batch["id"])
         state.remove_missing_manual_items(batch["id"], seen)
 
 
@@ -581,27 +632,33 @@ async def _import_ready_sab_jobs(importer: Importer, force: bool = False) -> int
     return imported
 
 
-def _import_queue_item(item: dict[str, object], importer: Importer) -> int:
+def _import_queue_item(item: dict[str, object], importer: Importer, *, force: bool = False, from_worker: bool = False) -> int:
-    if item["source_type"] not in {"sab", "manual"} or item["state"] not in {"ready", "manual_batch", "failed"}:
+    if item["source_type"] not in {"sab", "manual"} or item["state"] not in {"ready", "manual_batch", "failed", "importing", "retrying", "waiting_for_sab"}:
         return 0
     source_path = item.get("source_path")
     if not source_path:
-        state.mark_queue_item(str(item["source_type"]), str(item["source_id"]), "failed", "missing source path")
+        state.mark_queue_item_result(str(item["source_type"]), str(item["source_id"]), "failed", "missing source path", increment_attempts=True)
         return 0
     source = Path(str(source_path))
+    if not source.exists():
+        state.mark_queue_item_result(str(item["source_type"]), str(item["source_id"]), "failed", "missing source file", increment_attempts=True)
+        return 0
+    if not from_worker:
+        state.mark_queue_item(str(item["source_type"]), str(item["source_id"]), "importing", item.get("reason"))
     started = set_current_job(str(source))
     try:
         result = importer.import_file(source, should_cancel=consume_cancel_request, on_progress=lambda copied, total: set_current_job(str(source), bytes_copied=copied, total_bytes=total, started_at=started))
         state.add_history(result.source, result.target, "imported", result.bytes)
-        state.mark_queue_item(str(item["source_type"]), str(item["source_id"]), "imported")
+        state.mark_queue_item_result(str(item["source_type"]), str(item["source_id"]), "imported", increment_attempts=True)
         return 1
     except ImportCancelled:
         state.add_history(source, source, "cancelled", 0, "cancelled")
-        state.mark_queue_item(str(item["source_type"]), str(item["source_id"]), "skipped", "cancelled")
+        state.mark_queue_item_result(str(item["source_type"]), str(item["source_id"]), "skipped", "cancelled", increment_attempts=True)
         return 0
     except Exception as exc:
         state.add_history(source, source, "failed", 0, exc.__class__.__name__)
-        state.mark_queue_item(str(item["source_type"]), str(item["source_id"]), "failed", exc.__class__.__name__)
+        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)
         return 0
     finally:
         set_current_job(None)
@@ -617,26 +674,22 @@ def _import_manual_batches(importer: Importer) -> int:
         for item in items:
             if consume_cancel_request():
                 return imported
-            source = Path(item["source_path"])
+            imported += _import_queue_item(item, importer, force=True)
-            started = set_current_job(str(source))
-            try:
-                result = importer.import_file(source, should_cancel=consume_cancel_request, on_progress=lambda copied, total, source=source, started=started: set_current_job(str(source), bytes_copied=copied, total_bytes=total, started_at=started))
-                state.add_history(result.source, result.target, "imported", result.bytes)
-                state.mark_queue_item("manual", item["source_id"], "imported")
-                imported += 1
-            except ImportCancelled:
-                state.add_history(source, source, "cancelled", 0, "cancelled")
-                state.mark_queue_item("manual", item["source_id"], "skipped", "cancelled")
-                return imported
-            except Exception as exc:
-                state.add_history(source, source, "failed", 0, exc.__class__.__name__)
-                state.mark_queue_item("manual", item["source_id"], "failed", exc.__class__.__name__)
-            finally:
-                set_current_job(None)
         if not scan_videos(path):
             state.complete_manual_batch(batch["id"])
     return imported
 
 
+@app.on_event("startup")
+def startup_queue_worker() -> None:
+    ensure_worker_running()
+
+
+@app.on_event("shutdown")
+def shutdown_queue_worker() -> None:
+    stop_worker()
+    state.release_stale_claims(WORKER_ID)
+
+
 def run() -> None:
     uvicorn.run("importarr.main:app", host=settings.bind_host, port=settings.bind_port, reload=False)

importarr/state.py

@@ -5,6 +5,17 @@ from pathlib import Path
 from typing import Any
 
 
+ACTIVE_QUEUE_STATES = {
+    "detected",
+    "waiting_for_sab",
+    "ready",
+    "importing",
+    "retrying",
+}
+
+TERMINAL_QUEUE_STATES = {"imported", "failed", "skipped"}
+
+
 class State:
     def __init__(self, path: Path):
         self.path = path
@@ -47,6 +58,10 @@ class State:
               batch_id integer,
               job_id text,
               sab_category text,
+              attempt_count integer not null default 0,
+              last_error text,
+              claimed_by text,
+              claimed_at text,
               first_seen_at text not null default current_timestamp,
               updated_at text not null default current_timestamp,
               completed_at text,
@@ -57,6 +72,14 @@ class State:
         columns = {row["name"] for row in self.conn.execute("pragma table_info(import_queue_items)")}
         if "sab_category" not in columns:
             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 "last_error" not in columns:
+            self.conn.execute("alter table import_queue_items add column last_error text")
+        if "claimed_by" not in columns:
+            self.conn.execute("alter table import_queue_items add column claimed_by text")
+        if "claimed_at" not in columns:
+            self.conn.execute("alter table import_queue_items add column claimed_at text")
         self.conn.commit()
 
     def get_app_state(self, key: str, default: str | None = None) -> str | None:
@@ -116,6 +139,7 @@ class State:
         batch_id: int | None = None,
         job_id: str | None = None,
         sab_category: str | None = None,
+        preserve_finished_state: bool = True,
     ) -> dict[str, Any]:
         self.conn.execute(
             """
@@ -124,17 +148,27 @@ class State:
             on conflict(source_type, source_id) do update set
               source_path=excluded.source_path,
               name=excluded.name,
-              state=excluded.state,
+              state=case
-              reason=excluded.reason,
+                when ? and import_queue_items.state in ('imported','failed','skipped','importing','retrying') then import_queue_items.state
+                else excluded.state
+              end,
+              reason=case
+                when ? and import_queue_items.state in ('imported','failed','skipped','importing','retrying') then coalesce(import_queue_items.reason, excluded.reason)
+                else excluded.reason
+              end,
               relative_path=excluded.relative_path,
               size=excluded.size,
               batch_id=excluded.batch_id,
               job_id=excluded.job_id,
               sab_category=excluded.sab_category,
               updated_at=current_timestamp,
-              completed_at=case when excluded.state in ('imported','failed','skipped') then current_timestamp else null 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
+                else null
+              end
             """,
-            (source_type, source_id, str(source_path) if source_path else None, name, state, reason, relative_path, size, batch_id, job_id, sab_category),
+            (source_type, source_id, str(source_path) if source_path else None, name, state, reason, relative_path, size, batch_id, job_id, sab_category, preserve_finished_state, preserve_finished_state, preserve_finished_state),
         )
         self.conn.commit()
         row = self.conn.execute("select * from import_queue_items where source_type = ? and source_id = ?", (source_type, source_id)).fetchone()
@@ -142,11 +176,76 @@ class State:
 
     def mark_queue_item(self, source_type: str, source_id: str, state: str, reason: str | None = None) -> None:
         self.conn.execute(
-            "update import_queue_items set state=?, reason=?, updated_at=current_timestamp, 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, 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),
         )
         self.conn.commit()
 
+    def claim_next_queue_item(self, worker_id: str) -> dict[str, Any] | None:
+        with self.conn:
+            row = self.conn.execute(
+                """
+                select * from import_queue_items
+                where state in ('ready','retrying') and claimed_by is null
+                order by
+                  case state when 'ready' then 0 else 1 end,
+                  updated_at asc,
+                  id asc
+                limit 1
+                """
+            ).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, row["id"]),
+            )
+            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:
+        if worker_id is None:
+            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 mark_queue_item_result(
+        self,
+        source_type: str,
+        source_id: str,
+        state: str,
+        reason: str | None = None,
+        *,
+        increment_attempts: bool = False,
+    ) -> None:
+        self.conn.execute(
+            """
+            update import_queue_items
+            set state=?,
+                reason=?,
+                last_error=case when ? in ('failed','retrying','skipped') then ? else null end,
+                attempt_count=attempt_count + ?,
+                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),
+        )
+        self.conn.commit()
+
     def delete_queue_item(self, item_id: int) -> bool:
         cursor = self.conn.execute("delete from import_queue_items where id = ?", (item_id,))
         self.conn.commit()

importarr/templates/index.html

@@ -39,6 +39,7 @@
       <article><strong>{{ status.current or 'Idle' }}</strong><span>Current import</span></article>
       <article><strong>{{ status.imported_total }}</strong><span>Imported</span></article>
       <article><strong>{{ status.failed_total }}</strong><span>Failed</span></article>
+      <article><strong>{{ status.queue_total }}</strong><span>Queue items</span></article>
     </section>
     <section class="panel current-panel" aria-label="Current import details">
       <div class="section-title"><h2>Current import</h2><span id="current-runtime">idle</span></div>
@@ -131,7 +132,7 @@
   function renderUpdateCheck(update){ const banner=document.getElementById('update-banner'); if(!update?.update_available){ banner.hidden=true; return; } document.getElementById('update-message').textContent=`↑ New version available: ${update.latest_version}`; document.getElementById('update-command').textContent=`current ${update.current_version}`; banner.hidden=false; }
   async function checkForUpdates(){ const response=await fetch('/api/control/update-check'); if(!response.ok) return; renderUpdateCheck(await response.json()); }
   function actionButtons(j){ const buttons=[]; if(j.can_run_now) buttons.push(`<button type="button" data-action="run-now" data-id="${j.id}" title="Run now">▶</button>`); if(j.can_retry) buttons.push(`<button type="button" data-action="retry" data-id="${j.id}" title="Retry">↻</button>`); if(j.can_ignore) buttons.push(`<button type="button" data-action="ignore" data-id="${j.id}" class="warn" title="Ignore">!</button>`); if(j.can_remove) buttons.push(`<button type="button" data-action="remove" data-id="${j.id}" class="danger" title="Remove">🗑</button>`); return buttons.join(' '); }
-  function jobSubtext(j){ return `${esc(j.source_type)}${j.batch_id?' · batch '+esc(j.batch_id):''} · ${esc(j.relative_path||j.storage||j.source_id)}`; }
+  function jobSubtext(j){ return `${esc(j.source_type)}${j.batch_id?' · batch '+esc(j.batch_id):''} · attempts ${esc(j.attempt_count||0)} · ${esc(j.relative_path||j.storage||j.source_id)}`; }
   function readiness(j){ return `<span class="state" title="${esc(j.reason||j.state)}">${esc(j.state)}</span>`; }
   const fmtBytes=value=>{ const bytes=Number(value||0); if(!bytes)return 'size unknown'; const units=['B','KB','MB','GB','TB']; let n=bytes,i=0; while(n>=1024&&i<units.length-1){n/=1024;i++;} return `${n.toFixed(n>=10||i===0?0:1)} ${units[i]}`; };
   const fmtDuration=value=>{ const s=Math.max(0,Math.floor(Number(value||0))); const m=Math.floor(s/60); const r=s%60; return m?`${m}m ${String(r).padStart(2,'0')}s`:`${r}s`; };

tests/test_queue_controls.py

@@ -106,6 +106,7 @@ def test_queue_jobs_include_groups_and_manual_context(tmp_path, monkeypatch):
     assert jobs[0]["group"] == "manual_batch"
     assert jobs[0]["relative_path"] == "Release/Season 1/Episode.mkv"
     assert jobs[0]["can_run_now"] is True
+    assert jobs[0]["state"] == "ready"
 
 
 def test_queue_item_retry_ignore_and_remove_actions(tmp_path, monkeypatch):
@@ -132,8 +133,8 @@ def test_queue_item_run_now_imports_only_selected_item(tmp_path, monkeypatch):
     waiting = batch / "Waiting.mkv"
     selected.write_bytes(b"selected")
     waiting.write_bytes(b"waiting")
-    selected_row = main.state.upsert_queue_item(source_type="manual", source_id=str(selected), source_path=selected, name=selected.name, state="manual_batch")
+    selected_row = main.state.upsert_queue_item(source_type="manual", source_id=str(selected), source_path=selected, name=selected.name, state="ready")
-    main.state.upsert_queue_item(source_type="manual", source_id=str(waiting), source_path=waiting, name=waiting.name, state="manual_batch")
+    main.state.upsert_queue_item(source_type="manual", source_id=str(waiting), source_path=waiting, name=waiting.name, state="ready")
 
     result = main.queue_item_action(selected_row["id"], main.QueueItemActionRequest(action="run-now"))
 
@@ -143,7 +144,7 @@ def test_queue_item_run_now_imports_only_selected_item(tmp_path, monkeypatch):
     assert waiting.exists()
     rows = {row["name"]: row for row in main.state.list_queue_items(active_only=False)}
     assert rows["Selected.mkv"]["state"] == "imported"
-    assert rows["Waiting.mkv"]["state"] == "manual_batch"
+    assert rows["Waiting.mkv"]["state"] == "ready"
 
 
 def test_current_job_status_includes_progress(tmp_path, monkeypatch):
@@ -198,6 +199,39 @@ def test_cancel_current_stops_active_copy(tmp_path, monkeypatch):
     assert main.state.list_queue_items(active_only=False)[0]["state"] == "skipped"
 
 
+def test_status_includes_queue_counts(tmp_path, monkeypatch):
+    main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
+    main.state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")
+    main.state.upsert_queue_item(source_type="manual", source_id="b", name="B.mkv", state="failed")
+
+    status = main.status()
+
+    assert status["queue_total"] == 2
+    assert status["queue_counts"]["ready"] == 1
+    assert status["queue_counts"]["failed"] == 1
+
+
+def test_worker_claimed_failure_retries_item(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")
+    claimed = main.state.claim_next_queue_item(main.WORKER_ID)
+
+    class BrokenImporter:
+        def import_file(self, *args, **kwargs):
+            raise RuntimeError("boom")
+
+    imported = main._import_queue_item(claimed, BrokenImporter(), from_worker=True)
+    updated = main.state.get_queue_item(row["id"])
+
+    assert imported == 0
+    assert updated["state"] == "retrying"
+    assert updated["attempt_count"] == 1
+
+
 def test_control_update_runs_configured_command(tmp_path, monkeypatch):
     main, _download, _movies, _tv = configure_main(tmp_path, monkeypatch)
     main.settings.update_command = ["upgrade", "now"]

tests/test_state.py

@@ -11,3 +11,37 @@ def test_delete_queue_items_by_state_can_target_reason(tmp_path):
     rows = state.list_queue_items()
     assert len(rows) == 1
     assert rows[0]["source_id"] == "other"
+
+
+def test_upsert_preserves_terminal_state_by_default(tmp_path):
+    state = State(tmp_path / "state.db")
+    state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="failed", reason="boom")
+
+    updated = state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready", reason="rescanned")
+
+    assert updated["state"] == "failed"
+    assert updated["reason"] == "boom"
+
+
+def test_claim_next_queue_item_marks_importing(tmp_path):
+    state = State(tmp_path / "state.db")
+    state.upsert_queue_item(source_type="manual", source_id="a", name="A.mkv", state="ready")
+
+    row = state.claim_next_queue_item("worker-1")
+
+    assert row is not None
+    assert row["state"] == "importing"
+    assert row["claimed_by"] == "worker-1"
+
+
+def test_release_stale_claims_requeues_importing_items(tmp_path):
+    state = State(tmp_path / "state.db")
+    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")
+    row = state.get_queue_item(1)
+
+    assert released == 1
+    assert row["state"] == "retrying"
+    assert row["claimed_by"] is None