diff --git a/importarr/main.py b/importarr/main.py
index b035dde..ff0a633 100644
--- a/importarr/main.py
+++ b/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"
- state.mark_queue_item(item["source_type"], item["source_id"], retry_state, "retry requested")
+ 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))
+ 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:
- if item["source_type"] not in {"sab", "manual"} or item["state"] not in {"ready", "manual_batch", "failed"}:
+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", "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"])
- 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)
+ imported += _import_queue_item(item, importer, force=True)
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)
diff --git a/importarr/state.py b/importarr/state.py
index 66402ec..aaeb1fb 100644
--- a/importarr/state.py
+++ b/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,
- reason=excluded.reason,
+ state=case
+ 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()
diff --git a/importarr/templates/index.html b/importarr/templates/index.html
index f7a1751..725c7df 100644
--- a/importarr/templates/index.html
+++ b/importarr/templates/index.html
@@ -39,6 +39,7 @@
{{ status.current or 'Idle' }}Current import
{{ status.imported_total }}Imported
{{ status.failed_total }}Failed
+ {{ status.queue_total }}Queue items
Current import
idle
@@ -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(``); if(j.can_retry) buttons.push(``); if(j.can_ignore) buttons.push(``); if(j.can_remove) buttons.push(``); 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 `${esc(j.state)}`; }
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=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`; };
diff --git a/tests/test_queue_controls.py b/tests/test_queue_controls.py
index 6406db9..dec7b24 100644
--- a/tests/test_queue_controls.py
+++ b/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")
- main.state.upsert_queue_item(source_type="manual", source_id=str(waiting), source_path=waiting, name=waiting.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="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"]
diff --git a/tests/test_state.py b/tests/test_state.py
index 3e13e62..2075ab1 100644
--- a/tests/test_state.py
+++ b/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