diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 5c363ec..457dba3 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -4,9 +4,9 @@ How `recoll-modern` is built today. Update in the same commit as any structural ## Overview -The Python service scaffold, SQLite schema, collection and membership repositories, and document -discovery/identity modules are built. The API, job queue, and Recoll adapter remain intended structure. -Recoll behaviour below was verified against a real container. +The Python service scaffold, SQLite schema, collection and membership repositories, document +discovery/identity modules, and durable job worker are built. The API and Recoll adapter remain +intended structure. Recoll behaviour below was verified against a real container. The goal: a self-hosted service that wraps Recoll's indexing and full-text search behind (a) a modern web UI and (b) an MCP server, running in Docker, with several independent collections such as @@ -24,6 +24,8 @@ web UI and (b) an MCP server, running in Docker, with several independent collec | `src/recoll_modern/storage/` | SQLite migrations, collection metadata and content-hash identity | | `src/recoll_modern/domain/membership.py` | Validates sources and applies direct additions and rule changes | | `src/recoll_modern/indexing/reconcile.py` | Refreshes persistent directory rules while retaining excluded or unreadable files as present | +| `src/recoll_modern/indexing/worker.py` | Claims one durable job at a time and runs blocking handlers off the event loop | +| `src/recoll_modern/indexing/handlers.py` | Scans membership sources before strict-mode mutations and records file errors | | `tests/` | Automated service tests | Discovery produces content hashes and root-relative paths without writing to the document volume. @@ -80,7 +82,9 @@ is normalized: `direct_memberships` holds explicit files and snapshot expansions `membership_exclusions` holds per-file overrides against rule-derived membership. Effective membership is direct membership union non-excluded rule membership, so direct additions remain additive. -`jobs` and `job_file_errors` make work and partial failures durable. `users` and hashed-token `sessions` +`jobs` and `job_file_errors` make work and partial failures durable. At startup, the worker requeues +well-formed interrupted jobs, clears errors from the prior attempt, and fails unreadable requests. +`users` and hashed-token `sessions` provide authentication. `schema_migrations` is forward-only. The exact columns and constraints are the contract in `docs/specs/001-search-engine-api.md`. diff --git a/docs/TODO.md b/docs/TODO.md index 1becf5c..e1ebfcc 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -15,11 +15,12 @@ Where the workflow stands (`AGENTS.md` -> Workflow: brainstorm -> spec -> plan - selectable search fields, subpath hosting and the packaging seam for deferred desktop spec 005. - Plan `docs/plans/001-search-engine-api.md`: **approved 2026-09-18.** Its listed Python runtime and development dependencies are approved. Execution follows the plan task by task. -- Plan Tasks 1–3 are complete and independently reviewed. Tasks 4–5 add read-only discovery, - content-hash identity, additive membership and directory rules. Their checkboxes are current in the - plan. The next task is the durable job queue. +- Plan Tasks 1–3 are complete and independently reviewed. Tasks 4–6 add read-only discovery, + content-hash identity, additive membership, directory rules and durable jobs. Their checkboxes are + current in the plan. The next task is membership-map export and the native helper. -The next action is Task 6, "Add the durable single-worker job queue." Do not repeat Tasks 1–5. +The next action is Task 7, "Export membership atomically and build the native helper." Do not repeat +Tasks 1–6. Things a resuming agent will otherwise get wrong: diff --git a/docs/plans/001-search-engine-api.md b/docs/plans/001-search-engine-api.md index 03fcec2..6242dfd 100644 --- a/docs/plans/001-search-engine-api.md +++ b/docs/plans/001-search-engine-api.md @@ -478,7 +478,7 @@ git commit -m "feat: add membership provenance and rules" - Produces: `JobWorker.start()`, `JobWorker.stop()`, `JobWorker.run_once() -> bool` - Produces: `MembershipJobHandler.handle(job: Job) -> None` -- [ ] **Step 1: Write failing transition, recovery, and policy tests** +- [x] **Step 1: Write failing transition, recovery, and policy tests** ```python def test_recover_running_requeues_valid_jobs(repo: JobRepository) -> None: @@ -498,17 +498,17 @@ def test_best_effort_applies_valid_files_and_records_errors(handler: MembershipJ assert jobs.current.state == JobState.COMPLETED_WITH_ERRORS ``` -- [ ] **Step 2: Run tests and observe failure** +- [x] **Step 2: Run tests and observe failure** Run: `uv run pytest tests/storage/test_job_repository.py tests/indexing/test_worker.py tests/indexing/test_job_handlers.py -v` -- [ ] **Step 3: Implement durable transitions and handlers** +- [x] **Step 3: Implement durable transitions and handlers** Use opaque `j` plus 32 hex digits. `claim_next` runs under `BEGIN IMMEDIATE`, selects the oldest queued job, and changes it to running atomically. Valid transitions are queued→running, running→completed/completed_with_errors/failed, and failed→queued via retry. Startup requeues running jobs whose JSON decodes to the registered request type; malformed requests become failed with `invalid_job_request`. The worker owns one asyncio task, sleeps on an `asyncio.Event`, performs blocking discovery/database/engine work with `asyncio.to_thread`, and on shutdown stops claiming new work and leaves an interrupted running job recoverable. Membership handlers scan completely before opening the desired-state transaction in strict mode. Best-effort handlers persist every safe file error and continue. -- [ ] **Step 4: Verify and commit** +- [x] **Step 4: Verify and commit** Run: `uv run pytest tests/storage/test_job_repository.py tests/indexing/test_worker.py tests/indexing/test_job_handlers.py -v && uv run ruff check . && uv run mypy src tests` diff --git a/src/recoll_modern/domain/jobs.py b/src/recoll_modern/domain/jobs.py new file mode 100644 index 0000000..2cff2a4 --- /dev/null +++ b/src/recoll_modern/domain/jobs.py @@ -0,0 +1,32 @@ +"""Durable indexing-job state shared by storage and workers.""" + +from dataclasses import dataclass +from enum import StrEnum + + +class JobState(StrEnum): + QUEUED = "queued" + RUNNING = "running" + COMPLETED = "completed" + COMPLETED_WITH_ERRORS = "completed_with_errors" + FAILED = "failed" + + +@dataclass(frozen=True, slots=True) +class Job: + id: str + kind: str + state: JobState + request: dict[str, object] + fail_on_file_error: bool + progress: int + total: int + error_code: str | None = None + error_message: str | None = None + + +@dataclass(frozen=True, slots=True) +class JobFileError: + relative_path: str + code: str + message: str diff --git a/src/recoll_modern/domain/membership.py b/src/recoll_modern/domain/membership.py index d86e163..52ae764 100644 --- a/src/recoll_modern/domain/membership.py +++ b/src/recoll_modern/domain/membership.py @@ -27,7 +27,16 @@ class MembershipService: self._memberships = memberships def plan_add(self, collection_id: str, sources: list[Source]) -> MembershipPlanResult: - for source in sources: + discovered = [(source, discover_sources(self._policy, [source])) for source in sources] + return self.apply_discovered(collection_id, discovered) + + def apply_discovered( + self, + collection_id: str, + discovered: list[tuple[Source, DiscoveryReport]], + ) -> MembershipPlanResult: + """Apply already scanned sources so strict jobs never rescan after validation.""" + for source, _ in discovered: if source.type == "directory": parent = self._policy.normalize(source.path) if any( @@ -35,7 +44,6 @@ class MembershipService: for path in source.excluded_directories ): raise ValueError("excluded directory must be a descendant") - discovered = [(source, discover_sources(self._policy, [source])) for source in sources] files = list({ file.relative_path: file for _, report in discovered diff --git a/src/recoll_modern/indexing/handlers.py b/src/recoll_modern/indexing/handlers.py new file mode 100644 index 0000000..b8d9b0c --- /dev/null +++ b/src/recoll_modern/indexing/handlers.py @@ -0,0 +1,71 @@ +"""Durable membership job application after complete source discovery.""" + +from recoll_modern.domain.jobs import Job, JobFileError +from recoll_modern.domain.membership import MembershipService +from recoll_modern.indexing.discovery import Source, discover_sources +from recoll_modern.indexing.paths import PathPolicy +from recoll_modern.storage.documents import DocumentRepository +from recoll_modern.storage.jobs import JobRepository +from recoll_modern.storage.membership import MembershipRepository + + +class MembershipJobHandler: + def __init__( + self, + policy: PathPolicy, + documents: DocumentRepository, + memberships: MembershipRepository, + jobs: JobRepository, + ) -> None: + self._policy = policy + self._service = MembershipService(policy, documents, memberships) + self._jobs = jobs + + def handle(self, job: Job) -> None: + try: + collection_id = job.request["collection_id"] + source_values = job.request["sources"] + if not isinstance(collection_id, str) or not isinstance(source_values, list): + raise ValueError + sources = [_parse_source(value) for value in source_values] + except (KeyError, ValueError, TypeError): + self._jobs.fail(job.id, "invalid_job_request", "Job request is invalid") + return + discovered = [(source, discover_sources(self._policy, [source])) for source in sources] + errors = [error for _, report in discovered for error in report.errors] + for error in errors: + self._jobs.add_file_error( + job.id, JobFileError(error.relative_path, error.code, "File could not be read") + ) + total = sum(len(report.files) + len(report.errors) for _, report in discovered) + self._jobs.progress(job.id, 0, total) + if errors and job.fail_on_file_error: + self._jobs.fail(job.id, "file_error", "One or more files could not be read") + return + try: + self._service.apply_discovered(collection_id, discovered) + except ValueError: + self._jobs.fail(job.id, "invalid_job_request", "Job request is invalid") + return + self._jobs.progress(job.id, total, total) + self._jobs.complete(job.id, with_errors=bool(errors)) + + +def _parse_source(value: object) -> Source: + if not isinstance(value, dict): + raise ValueError + kind = value.get("type") + path = value.get("path") + if not isinstance(path, str): + raise ValueError + if kind == "file": + return Source.file(path) + if kind == "directory": + mode = value.get("mode") + exclusions = value.get("excluded_directories", []) + if mode not in ("snapshot", "persistent") or not isinstance(exclusions, list): + raise ValueError + if not all(isinstance(item, str) for item in exclusions): + raise ValueError + return Source.directory(path, mode, tuple(exclusions)) + raise ValueError diff --git a/src/recoll_modern/indexing/worker.py b/src/recoll_modern/indexing/worker.py new file mode 100644 index 0000000..a33596a --- /dev/null +++ b/src/recoll_modern/indexing/worker.py @@ -0,0 +1,68 @@ +"""Single asynchronous worker around blocking durable job handlers.""" + +import asyncio +from collections.abc import Callable, Mapping + +from recoll_modern.domain.jobs import Job, JobState +from recoll_modern.storage.jobs import JobRepository + + +class JobWorker: + def __init__( + self, jobs: JobRepository, handlers: Mapping[str, Callable[[Job], None]] + ) -> None: + self._jobs = jobs + self._handlers = handlers + self._wake = asyncio.Event() + self._stopping = False + self._task: asyncio.Task[None] | None = None + + def start(self) -> None: + if self._task is not None: + return + self._jobs.recover_running() + self._stopping = False + self._task = asyncio.create_task(self._run()) + self._wake.set() + + async def stop(self) -> None: + self._stopping = True + self._wake.set() + if self._task is not None: + await self._task + self._task = None + + def notify(self) -> None: + self._wake.set() + + async def run_once(self) -> bool: + job = await asyncio.to_thread(self._jobs.claim_next) + if job is None: + return False + handler = self._handlers.get(job.kind) + if handler is None: + await asyncio.to_thread( + self._jobs.fail, job.id, "invalid_job_request", "Job kind is not supported" + ) + return True + try: + await asyncio.to_thread(handler, job) + except Exception: + await asyncio.to_thread( + self._jobs.fail, job.id, "job_failed", "Job failed during processing" + ) + else: + current = await asyncio.to_thread(self._jobs.get, job.id) + if current is not None and current.state is JobState.RUNNING: + await asyncio.to_thread(self._jobs.complete, job.id) + return True + + async def _run(self) -> None: + while not self._stopping: + if await self.run_once(): + continue + self._wake.clear() + try: + await asyncio.wait_for(self._wake.wait(), timeout=1.0) + except TimeoutError: + pass diff --git a/src/recoll_modern/storage/jobs.py b/src/recoll_modern/storage/jobs.py new file mode 100644 index 0000000..4dbb169 --- /dev/null +++ b/src/recoll_modern/storage/jobs.py @@ -0,0 +1,167 @@ +"""Atomic SQLite transitions for indexing jobs.""" + +import json +import secrets +import sqlite3 +from collections.abc import Mapping +from datetime import UTC, datetime + +from recoll_modern.domain.jobs import Job, JobFileError, JobState +from recoll_modern.storage.database import Database + + +class JobRepository: + def __init__(self, database: Database) -> None: + self._database = database + + def enqueue( + self, kind: str, request: Mapping[str, object], fail_on_file_error: bool + ) -> Job: + serialized = json.dumps(dict(request), sort_keys=True) + now = datetime.now(UTC).isoformat() + while True: + job_id = f"j{secrets.token_hex(16)}" + try: + with self._database.transaction() as connection: + connection.execute( + "INSERT INTO jobs " + "(id, kind, state, request_json, fail_on_file_error, created_at) " + "VALUES (?, ?, 'queued', ?, ?, ?)", + (job_id, kind, serialized, int(fail_on_file_error), now), + ) + except sqlite3.IntegrityError as error: + if "jobs.id" in str(error): + continue + raise + job = self.get(job_id) + assert job is not None + return job + + def get(self, job_id: str) -> Job | None: + with self._database.connect() as connection: + row = connection.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone() + return _job_from_row(row) if row is not None else None + + def claim_next(self) -> Job | None: + with self._database.transaction() as connection: + row = connection.execute( + "SELECT * FROM jobs WHERE state = 'queued' ORDER BY created_at, rowid LIMIT 1" + ).fetchone() + if row is None: + return None + now = datetime.now(UTC).isoformat() + connection.execute( + "UPDATE jobs SET state = 'running', started_at = ?, finished_at = NULL " + "WHERE id = ? AND state = 'queued'", + (now, row["id"]), + ) + updated = connection.execute( + "SELECT * FROM jobs WHERE id = ?", (row["id"],) + ).fetchone() + return _job_from_row(updated) + + def progress(self, job_id: str, progress: int, total: int) -> None: + with self._database.transaction() as connection: + connection.execute( + "UPDATE jobs SET progress = ?, total = ? " + "WHERE id = ? AND state = 'running'", + (progress, total, job_id), + ) + + def add_file_error(self, job_id: str, error: JobFileError) -> None: + with self._database.transaction() as connection: + connection.execute( + "INSERT OR REPLACE INTO job_file_errors " + "(job_id, relative_path, code, message) VALUES (?, ?, ?, ?)", + (job_id, error.relative_path, error.code, error.message), + ) + + def file_errors(self, job_id: str) -> list[JobFileError]: + with self._database.connect() as connection: + return [ + JobFileError(str(row[0]), str(row[1]), str(row[2])) + for row in connection.execute( + "SELECT relative_path, code, message FROM job_file_errors " + "WHERE job_id = ? ORDER BY relative_path", + (job_id,), + ) + ] + + def complete(self, job_id: str, *, with_errors: bool = False) -> None: + state = "completed_with_errors" if with_errors else "completed" + with self._database.transaction() as connection: + connection.execute( + "UPDATE jobs SET state = ?, finished_at = ? " + "WHERE id = ? AND state = 'running'", + (state, datetime.now(UTC).isoformat(), job_id), + ) + + def fail(self, job_id: str, code: str, message: str) -> None: + with self._database.transaction() as connection: + connection.execute( + "UPDATE jobs SET state = 'failed', error_code = ?, error_message = ?, " + "finished_at = ? WHERE id = ? AND state = 'running'", + (code, message, datetime.now(UTC).isoformat(), job_id), + ) + + def recover_running(self) -> None: + with self._database.transaction() as connection: + for row in connection.execute( + "SELECT id, kind, request_json FROM jobs WHERE state = 'running'" + ).fetchall(): + if _valid_request(str(row["kind"]), str(row["request_json"])): + connection.execute( + "DELETE FROM job_file_errors WHERE job_id = ?", (row["id"],) + ) + connection.execute( + "UPDATE jobs SET state = 'queued', started_at = NULL, " + "progress = 0, total = 0 WHERE id = ?", + (row["id"],), + ) + else: + connection.execute( + "UPDATE jobs SET state = 'failed', error_code = 'invalid_job_request', " + "error_message = 'Job request is invalid', finished_at = ? WHERE id = ?", + (datetime.now(UTC).isoformat(), row["id"]), + ) + + def retry(self, job_id: str) -> None: + with self._database.transaction() as connection: + cursor = connection.execute( + "UPDATE jobs SET state = 'queued', error_code = NULL, error_message = NULL, " + "started_at = NULL, finished_at = NULL, progress = 0, total = 0 " + "WHERE id = ? AND state = 'failed'", + (job_id,), + ) + if cursor.rowcount != 1: + raise ValueError("job is not retryable") + connection.execute("DELETE FROM job_file_errors WHERE job_id = ?", (job_id,)) + + +def _valid_request(kind: str, raw: str) -> bool: + try: + value = json.loads(raw) + except (json.JSONDecodeError, TypeError): + return False + if not isinstance(value, dict): + return False + if kind == "membership_add": + return isinstance(value.get("collection_id"), str) and isinstance( + value.get("sources"), list + ) + return bool(kind) + + +def _job_from_row(row: sqlite3.Row) -> Job: + try: + value = json.loads(str(row["request_json"])) + except json.JSONDecodeError: + value = {} + request = value if isinstance(value, dict) else {} + return Job( + id=str(row["id"]), kind=str(row["kind"]), state=JobState(str(row["state"])), + request=request, fail_on_file_error=bool(row["fail_on_file_error"]), + progress=int(row["progress"]), total=int(row["total"]), + error_code=str(row["error_code"]) if row["error_code"] is not None else None, + error_message=str(row["error_message"]) if row["error_message"] is not None else None, + ) diff --git a/src/recoll_modern/storage/membership.py b/src/recoll_modern/storage/membership.py index 4e95288..0d7ea06 100644 --- a/src/recoll_modern/storage/membership.py +++ b/src/recoll_modern/storage/membership.py @@ -138,6 +138,16 @@ class MembershipRepository: (collection_id, content_hash), ).fetchone() is not None + def direct_documents(self, collection_id: str) -> set[str]: + with self._database.connect() as connection: + return { + str(row[0]) + for row in connection.execute( + "SELECT document_hash FROM direct_memberships WHERE collection_id = ?", + (collection_id,), + ) + } + def exclude_document(self, collection_id: str, content_hash: str) -> None: with self._database.transaction() as connection: connection.execute( diff --git a/tests/indexing/test_job_handlers.py b/tests/indexing/test_job_handlers.py new file mode 100644 index 0000000..6503b46 --- /dev/null +++ b/tests/indexing/test_job_handlers.py @@ -0,0 +1,58 @@ +from pathlib import Path + +from recoll_modern.domain.jobs import JobState +from recoll_modern.domain.models import CollectionCreate +from recoll_modern.indexing.handlers import MembershipJobHandler +from recoll_modern.indexing.paths import PathPolicy +from recoll_modern.storage.collections import SQLiteCollectionRepository +from recoll_modern.storage.database import Database +from recoll_modern.storage.documents import DocumentRepository +from recoll_modern.storage.jobs import JobRepository +from recoll_modern.storage.membership import MembershipRepository + + +def _setup( + tmp_path: Path, +) -> tuple[Path, JobRepository, MembershipRepository, MembershipJobHandler, str]: + root = tmp_path / "docs" + root.mkdir() + (root / "good.txt").write_text("good") + database = Database(tmp_path / "state.sqlite3") + database.migrate() + collection_id = SQLiteCollectionRepository(database).create(CollectionCreate("Books")).id + jobs = JobRepository(database) + membership = MembershipRepository(database) + handler = MembershipJobHandler( + PathPolicy(root), DocumentRepository(database), membership, jobs + ) + return root, jobs, membership, handler, collection_id + + +def _request(collection_id: str) -> dict[str, object]: + return {"collection_id": collection_id, "sources": [ + {"type": "file", "path": "good.txt"}, + {"type": "file", "path": "gone.txt"}, + ]} + + +def test_strict_discovery_error_applies_nothing(tmp_path: Path) -> None: + _, jobs, membership, handler, collection_id = _setup(tmp_path) + job = jobs.enqueue("membership_add", _request(collection_id), True) + claimed = jobs.claim_next() + assert claimed is not None + handler.handle(claimed) + assert jobs.get(job.id).state == JobState.FAILED # type: ignore[union-attr] + assert membership.direct_documents(collection_id) == set() + + +def test_best_effort_applies_valid_files_and_records_errors(tmp_path: Path) -> None: + _, jobs, membership, handler, collection_id = _setup(tmp_path) + job = jobs.enqueue("membership_add", _request(collection_id), False) + claimed = jobs.claim_next() + assert claimed is not None + handler.handle(claimed) + assert jobs.get(job.id).state == JobState.COMPLETED_WITH_ERRORS # type: ignore[union-attr] + assert len(membership.direct_documents(collection_id)) == 1 + assert [(error.relative_path, error.code) for error in jobs.file_errors(job.id)] == [ + ("gone.txt", "file_vanished") + ] diff --git a/tests/indexing/test_worker.py b/tests/indexing/test_worker.py new file mode 100644 index 0000000..60d3830 --- /dev/null +++ b/tests/indexing/test_worker.py @@ -0,0 +1,63 @@ +import asyncio +from pathlib import Path + +import pytest + +from recoll_modern.domain.jobs import Job, JobState +from recoll_modern.indexing.worker import JobWorker +from recoll_modern.storage.database import Database +from recoll_modern.storage.jobs import JobRepository + + +@pytest.mark.asyncio +async def test_worker_runs_one_job_and_records_completion(tmp_path: Path) -> None: + database = Database(tmp_path / "state.sqlite3") + database.migrate() + jobs = JobRepository(database) + job = jobs.enqueue("test", {}, False) + handled: list[str] = [] + + def handle(claimed: Job) -> None: + handled.append(claimed.id) + + worker = JobWorker(jobs, {"test": handle}) + assert await worker.run_once() + assert handled == [job.id] + assert jobs.get(job.id).state == JobState.COMPLETED # type: ignore[union-attr] + + +@pytest.mark.asyncio +async def test_worker_failure_is_durable(tmp_path: Path) -> None: + database = Database(tmp_path / "state.sqlite3") + database.migrate() + jobs = JobRepository(database) + job = jobs.enqueue("test", {}, False) + + def fail(_claimed: Job) -> None: + raise RuntimeError("private filesystem path") + + worker = JobWorker(jobs, {"test": fail}) + assert await worker.run_once() + failed = jobs.get(job.id) + assert failed is not None and failed.state == JobState.FAILED + assert "private" not in (failed.error_message or "") + + +@pytest.mark.asyncio +async def test_worker_start_recovers_and_drains_queue(tmp_path: Path) -> None: + database = Database(tmp_path / "state.sqlite3") + database.migrate() + jobs = JobRepository(database) + job = jobs.enqueue("test", {}, False) + jobs.claim_next() + handled: list[str] = [] + worker = JobWorker(jobs, {"test": lambda claimed: handled.append(claimed.id)}) + worker.start() + for _ in range(100): + current = jobs.get(job.id) + if current is not None and current.state == JobState.COMPLETED: + break + await asyncio.sleep(0.01) + await worker.stop() + assert handled == [job.id] + assert jobs.get(job.id).state == JobState.COMPLETED # type: ignore[union-attr] diff --git a/tests/storage/test_job_repository.py b/tests/storage/test_job_repository.py new file mode 100644 index 0000000..6ffc90e --- /dev/null +++ b/tests/storage/test_job_repository.py @@ -0,0 +1,70 @@ +from pathlib import Path + +from recoll_modern.domain.jobs import JobFileError, JobState +from recoll_modern.storage.database import Database +from recoll_modern.storage.jobs import JobRepository + + +def _repo(tmp_path: Path) -> JobRepository: + database = Database(tmp_path / "state.sqlite3") + database.migrate() + return JobRepository(database) + + +def test_claim_next_is_durable_and_oldest_first(tmp_path: Path) -> None: + repo = _repo(tmp_path) + first = repo.enqueue("membership_add", {"collection_id": "c1", "sources": []}, False) + second = repo.enqueue("membership_add", {"collection_id": "c1", "sources": []}, False) + assert repo.claim_next().id == first.id # type: ignore[union-attr] + assert repo.claim_next().id == second.id # type: ignore[union-attr] + assert repo.claim_next() is None + + +def test_recover_running_requeues_valid_jobs(tmp_path: Path) -> None: + repo = _repo(tmp_path) + job = repo.enqueue("membership_add", {"collection_id": "c1", "sources": []}, False) + repo.claim_next() + repo.recover_running() + assert repo.get(job.id).state == JobState.QUEUED # type: ignore[union-attr] + + +def test_recover_running_fails_malformed_request(tmp_path: Path) -> None: + database = Database(tmp_path / "state.sqlite3") + database.migrate() + repo = JobRepository(database) + job = repo.enqueue("membership_add", {"collection_id": "c1", "sources": []}, False) + repo.claim_next() + with database.transaction() as connection: + connection.execute("UPDATE jobs SET request_json = ? WHERE id = ?", ("{", job.id)) + repo.recover_running() + recovered = repo.get(job.id) + assert recovered is not None and recovered.state == JobState.FAILED + assert recovered.error_code == "invalid_job_request" + + +def test_recover_running_fails_invalid_source_shape(tmp_path: Path) -> None: + database = Database(tmp_path / "state.sqlite3") + database.migrate() + repo = JobRepository(database) + job = repo.enqueue("membership_add", {"collection_id": "c1", "sources": "not a list"}, False) + repo.claim_next() + repo.recover_running() + assert repo.get(job.id).state == JobState.FAILED # type: ignore[union-attr] + + +def test_recover_running_clears_errors_from_interrupted_attempt(tmp_path: Path) -> None: + repo = _repo(tmp_path) + job = repo.enqueue("membership_add", {"collection_id": "c1", "sources": []}, False) + repo.claim_next() + repo.add_file_error(job.id, JobFileError("gone.txt", "file_vanished", "File vanished")) + repo.recover_running() + assert repo.file_errors(job.id) == [] + + +def test_failed_job_can_be_retried(tmp_path: Path) -> None: + repo = _repo(tmp_path) + job = repo.enqueue("membership_add", {"collection_id": "c1", "sources": []}, False) + repo.claim_next() + repo.fail(job.id, "engine_unavailable", "Engine unavailable") + repo.retry(job.id) + assert repo.get(job.id).state == JobState.QUEUED # type: ignore[union-attr]