From 653c32db43bb811ddc6f21700fc82d7c485d0b63 Mon Sep 17 00:00:00 2001 From: Lucas Winther Date: Sun, 27 Sep 2026 19:12:56 +0200 Subject: [PATCH] fix: make indexing review paths replay-safe --- docs/ARCHITECTURE.md | 9 +- docs/DECISIONS.md | 16 +++ docs/TODO.md | 20 ++- src/recoll_modern/api/collections.py | 5 +- src/recoll_modern/api/documents.py | 115 ++++++++++++++-- src/recoll_modern/api/memberships.py | 8 +- src/recoll_modern/app.py | 33 ++++- src/recoll_modern/domain/membership.py | 2 +- src/recoll_modern/indexing/reconcile.py | 21 ++- src/recoll_modern/storage/collections.py | 70 ++++++---- src/recoll_modern/storage/jobs.py | 90 +++++++++++-- src/recoll_modern/storage/membership.py | 20 +++ tests/api/test_collections.py | 23 +++- tests/api/test_documents.py | 2 + tests/api/test_jobs.py | 38 ++++++ tests/api/test_memberships.py | 89 +++++++++++++ tests/contract/test_recoll_engine.py | 5 +- tests/indexing/test_job_handlers.py | 33 +++++ tests/indexing/test_reconcile.py | 30 +++++ tests/indexing/test_worker.py | 4 +- tests/integration/test_end_to_end.py | 145 +++++++++++++++++++++ tests/integration/test_read_only_corpus.py | 11 +- tests/security/test_downloads.py | 17 +++ tests/storage/test_job_repository.py | 39 +++++- 24 files changed, 759 insertions(+), 86 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index c146a3a..bc45153 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -83,7 +83,9 @@ Only `engine/` may import Recoll or invoke `recollindex`. Membership jobs first state in SQLite, atomically export the helper's `RCM1` flat map, then ask the engine to force-index affected canonical paths. Map records use length-prefixed absolute UTF-8 paths, so filenames can contain newlines. The C helper reads this private map without starting Python. Engine retries are -therefore idempotent. Persistent directory rules reconcile at startup and on an explicit reindex, +therefore idempotent. Jobs save paths needed for replay before changing rule membership, and collection +deletion commits the cleanup job in the same SQLite transaction. Persistent directory rules reconcile +at startup and on an explicit reindex, preserving prior membership for unreadable files or subtrees, not through live filesystem monitoring. ### Storage model @@ -114,8 +116,9 @@ derives effective membership from SQLite and includes direct and rule provenance Search passes all requested collection IDs into one engine query, then verifies hit paths and effective membership in SQLite before returning them. Full-text and original-file routes resolve a live canonical -path under the configured root. Original responses use range-capable `FileResponse`, with inline display -limited to safe media types. +path under the configured root. Original downloads open a contained regular file through no-follow +directory descriptors and stream from the pinned descriptor, including single byte ranges; inline +display is limited to safe media types. The image installs Recoll and its Python binding from Debian trixie, and syncs the application into a system-site-packages venv so the binding remains importable. A separate build stage compiles the C diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index dca4686..fcda202 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -3,6 +3,22 @@ Append-only, newest first. Never edit or delete an entry — supersede it with a new one that links back. One entry per decision a future maintainer would otherwise have to reverse-engineer. +## 2026-09-27 — Persist replay paths and pin original downloads +**Status:** accepted +**Context:** Review found that a job could mutate rule or collection state before its indexing step +failed, leaving a retry without the old paths. `FileResponse` also reopened a path after containment +checking, allowing a concurrent host-side symlink swap. +**Decision:** persist affected paths in a running job before rule mutations; enqueue collection cleanup +atomically with collection deletion. Open original files beneath the document root with no-follow +directory descriptors and stream from that checked descriptor, including byte-range responses. +**Alternatives:** recompute old paths on retry (the database may already have lost them); retain +`FileResponse` after a path-only check (it reopens the path). +**Consequences:** interrupted jobs retain their required reindex paths. File responses are tied to an +opened regular file even if directory entries change later. The descriptor traversal relies on Linux +`openat` semantics supplied by the current Docker deployment; desktop packaging must revisit it. + +--- + ## 2026-09-18 — Cross-platform desktop packaging is spec 005 **Status:** accepted **Supersedes:** "The product is four specs, built in order" (below, same date), only by appending a fifth diff --git a/docs/TODO.md b/docs/TODO.md index 0d02803..9f04248 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -5,7 +5,7 @@ Small, specific, independently actionable. Remove only when actually done. ## Resume here -Last touched 2026-09-26. Spec 001 implementation is in progress. +Last touched 2026-09-27. Spec 001 implementation is in review closure. Where the workflow stands (`AGENTS.md` -> Workflow: brainstorm -> spec -> plan -> TDD -> review): @@ -19,9 +19,19 @@ Where the workflow stands (`AGENTS.md` -> Workflow: brainstorm -> spec -> plan - identity, additive membership, durable jobs, the atomic membership map, the engine boundary, session authentication, the HTTP shell, membership/rule/job routes, scoped search and document retrieval, Docker packaging and acceptance coverage. Their checkboxes are current in the plan. - Final code review and branch handoff remain. + The first code review found replay, reconciliation, and download races. A fix checkpoint is on + `feat/001-document-discovery`; final verification and branch handoff remain. -The next action is Task 14 Step 6, "Request review before integration." Do not repeat Tasks 1–14. +The next action is Task 14 Step 6: run the full verification matrix, request a follow-up review of the +fixes, then hand off the branch. Do not repeat Tasks 1–14. The review fixes include idempotent rule +creation, persisted replay paths, atomic collection cleanup enqueue, unreadable-file preservation, +departed-path reindexing, directory-rule pagination, recovery request validation, and descriptor-pinned +downloads. The generated real-container lifecycle test covers these flows through Compose. Its 9-test +contract/integration suite passed with `RECOLL_MODERN_REQUIRE_DOCKER=1` on 2026-09-27; the host suite +passed with 157 passed and 4 skipped, and lint/type checking passed. The complete final matrix and +follow-up review still need to run before completion. Docker Compose may be running on +port 9124 with `RECOLL_MODERN_DOCUMENTS=/tmp/recoll-modern-corpus`; preserve its named volumes and +document data. Use `docker compose down` without `-v` when done. Things a resuming agent will otherwise get wrong: @@ -70,10 +80,6 @@ Things a resuming agent will otherwise get wrong: paths exist, and fail if `AGENTS.md` exceeds 150 lines. ## Known issues -- Original file download checks containment before returning `FileResponse`; harden against a local - symlink swap between that check and the response's file open before shipping to untrusted local - writers. The Compose document mount is read-only to the service, but host writers may still replace - entries concurrently. - The API test suite reports deprecation warnings from the installed FastAPI/Starlette TestClient dependency. Replace its client integration when the dependency stack supports the successor API. - Directory sources currently require a named child of the document root; decide whether the root diff --git a/src/recoll_modern/api/collections.py b/src/recoll_modern/api/collections.py index 836a42b..c5135c4 100644 --- a/src/recoll_modern/api/collections.py +++ b/src/recoll_modern/api/collections.py @@ -78,9 +78,8 @@ def delete_collection( collection = request.app.state.collections.get(collection_id) if confirmation != collection.name: raise AppError("confirmation_required", "Collection name confirmation is required", 409) - paths = request.app.state.collections.delete(collection_id) - job = request.app.state.jobs.enqueue( - "collection_cleanup", {"version": 1, "paths": paths}, False + job = request.app.state.collection_repository.delete_with_cleanup( + collection_id, request.app.state.jobs ) request.app.state.worker.notify() return {"job_id": job.id} diff --git a/src/recoll_modern/api/documents.py b/src/recoll_modern/api/documents.py index ef85b1e..2c5bfe0 100644 --- a/src/recoll_modern/api/documents.py +++ b/src/recoll_modern/api/documents.py @@ -1,11 +1,16 @@ """Complete extracted text and safe original-file responses.""" import mimetypes +import os +import re +import stat +from collections.abc import Iterator from pathlib import Path -from typing import cast +from typing import BinaryIO, cast +from urllib.parse import quote from fastapi import APIRouter, Request -from fastapi.responses import FileResponse +from fastapi.responses import Response, StreamingResponse from recoll_modern.api.dependencies import require_user from recoll_modern.domain.errors import AppError @@ -16,6 +21,7 @@ SAFE_INLINE_TYPES = frozenset({ "text/plain", "application/pdf", "image/png", "image/jpeg", "image/gif", "audio/mpeg", "audio/ogg", "video/mp4", }) +_RANGE = re.compile(r"bytes=(\d*)-(\d*)\Z") def _path(request: Request, document_hash: str) -> Path: @@ -31,6 +37,67 @@ def _path(request: Request, document_hash: str) -> Path: return cast(Path, path) +def open_contained_file(root: Path, relative_path: str) -> BinaryIO: + """Open beneath root before constructing a response; pin the checked inode.""" + root = root.resolve() + resolved = (root / relative_path).resolve(strict=True) + if not resolved.is_relative_to(root): + raise ValueError("file is outside document root") + parts = resolved.relative_to(root).parts + if not parts: + raise ValueError("document path is not a file") + directory_fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY) + try: + for part in parts[:-1]: + next_fd = os.open( + part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory_fd + ) + os.close(directory_fd) + directory_fd = next_fd + file_fd = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW, dir_fd=directory_fd) + if not stat.S_ISREG(os.fstat(file_fd).st_mode): + os.close(file_fd) + raise ValueError("document path is not a regular file") + return os.fdopen(file_fd, "rb") + finally: + os.close(directory_fd) + + +def _byte_range(header: str | None, size: int) -> tuple[int, int, bool] | None: + if header is None: + return (0, size - 1, False) + match = _RANGE.fullmatch(header.strip()) + if match is None or size == 0: + return None + start_text, end_text = match.groups() + if not start_text and not end_text: + return None + if not start_text: + suffix = int(end_text) + if suffix <= 0: + return None + return (max(size - suffix, 0), size - 1, True) + start = int(start_text) + end = min(int(end_text), size - 1) if end_text else size - 1 + if start >= size or end < start: + return None + return (start, end, True) + + +def _chunks(stream: BinaryIO, start: int, length: int) -> Iterator[bytes]: + try: + stream.seek(start) + remaining = length + while remaining: + data = stream.read(min(64 * 1024, remaining)) + if not data: + break + remaining -= len(data) + yield data + finally: + stream.close() + + @router.get("/documents/{document_hash}/text", name="document_text") def document_text(document_hash: str, request: Request) -> dict[str, str]: require_user(request) @@ -42,12 +109,44 @@ def document_text(document_hash: str, request: Request) -> dict[str, str]: @router.get("/documents/{document_hash}/file", name="document_file") -def document_file(document_hash: str, request: Request) -> FileResponse: +def document_file(document_hash: str, request: Request) -> Response: require_user(request) - path = _path(request, document_hash) - filename = path.name.replace("\r", "_").replace("\n", "_").replace('"', "_") + document = request.app.state.documents.get(document_hash) + if document is None: + raise AppError("document_not_found", "Document was not found", 404) + if document.canonical_path is None: + raise AppError("document_missing", "Document is missing", 410) + try: + stream = open_contained_file( + request.app.state.path_policy.root, document.canonical_path + ) + except (OSError, ValueError): + raise AppError("document_missing", "Document is missing", 410) from None + filename = Path(document.canonical_path).name + filename = filename.replace("\r", "_").replace("\n", "_").replace('"', "_") media_type = mimetypes.guess_type(filename)[0] or "application/octet-stream" - return FileResponse( - path, media_type=media_type, filename=filename, - content_disposition_type="inline" if media_type in SAFE_INLINE_TYPES else "attachment", + disposition = "inline" if media_type in SAFE_INLINE_TYPES else "attachment" + quoted = quote(filename) + content_disposition = ( + f'{disposition}; filename="{filename}"' + if quoted == filename else f"{disposition}; filename*=utf-8''{quoted}" + ) + size = os.fstat(stream.fileno()).st_size + selected = _byte_range(request.headers.get("range"), size) + if selected is None: + stream.close() + return Response(status_code=416, headers={ + "Accept-Ranges": "bytes", "Content-Range": f"bytes */{size}", + }) + start, end, partial = selected + headers = { + "Accept-Ranges": "bytes", + "Content-Length": str(end - start + 1), + "Content-Disposition": content_disposition, + } + if partial: + headers["Content-Range"] = f"bytes {start}-{end}/{size}" + return StreamingResponse( + _chunks(stream, start, end - start + 1), + status_code=206 if partial else 200, media_type=media_type, headers=headers, ) diff --git a/src/recoll_modern/api/memberships.py b/src/recoll_modern/api/memberships.py index 5f89f50..249279c 100644 --- a/src/recoll_modern/api/memberships.py +++ b/src/recoll_modern/api/memberships.py @@ -74,14 +74,18 @@ def remove_document( @router.get("/collections/{collection_id}/directory-rules") -def list_rules(collection_id: str, request: Request) -> dict[str, object]: +def list_rules( + collection_id: str, request: Request, limit: int = 50, offset: int = 0 +) -> dict[str, object]: require_user(request) request.app.state.collections.get(collection_id) + if not 1 <= limit <= 200 or offset < 0: + raise AppError("validation_failed", "Invalid pagination", 422) rules = [ asdict(rule) for rule in request.app.state.memberships.list_rules() if rule.collection_id == collection_id ] - return {"items": rules, "total": len(rules)} + return {"items": rules[offset:offset + limit], "total": len(rules)} @router.patch("/collections/{collection_id}/directory-rules/{rule_id}", status_code=202) diff --git a/src/recoll_modern/app.py b/src/recoll_modern/app.py index aa676b8..a0f6de9 100644 --- a/src/recoll_modern/app.py +++ b/src/recoll_modern/app.py @@ -66,6 +66,14 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) - if paths: engine.force_reindex(tuple(str(policy.root / path) for path in sorted(set(paths)))) + def current_rule_paths(rule_id: str) -> tuple[str, ...]: + paths = [] + for content_hash in memberships.rule_documents(rule_id): + record = documents.get(content_hash) + if record is not None and record.canonical_path is not None: + paths.append(record.canonical_path) + return tuple(paths) + def cleanup(job: Job) -> None: export_membership_map(map_rows, settings.membership_map_path) paths = job.request.get("paths", []) @@ -98,26 +106,39 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) - raise ValueError("invalid rule patch job") if not all(isinstance(item, str) for item in exclusions): raise ValueError("invalid rule patch job") + replay_paths = jobs.record_reindex_paths(job.id, current_rule_paths(rule_id)) result = membership_service.plan_patch_rule(rule_id, tuple(exclusions)) - reindex_changed(result.changed_paths) + reindex_changed(tuple(set(replay_paths) | set(result.changed_paths))) def delete_rule(job: Job) -> None: rule_id = job.request.get("rule_id") if not isinstance(rule_id, str): raise ValueError("invalid rule deletion job") - reindex_changed(membership_service.plan_delete_rule(rule_id)) + replay_paths = jobs.record_reindex_paths(job.id, current_rule_paths(rule_id)) + changed = membership_service.plan_delete_rule(rule_id) + reindex_changed(tuple(set(replay_paths) | set(changed))) def reindex(job: Job) -> None: paths = job.request.get("paths", []) collections_requested = job.request.get("collections", []) if not isinstance(paths, list) or not isinstance(collections_requested, list): raise ValueError("invalid reindex job") - for rule in memberships.list_rules(): - reconcile_rule(rule.id, policy, documents, memberships) + rules = memberships.list_rules() + prior_paths = tuple( + path for rule in rules for path in current_rule_paths(rule.id) + ) + replay_paths = jobs.record_reindex_paths(job.id, prior_paths) + changed: set[str] = set() + for rule in rules: + changed.update( + reconcile_rule(rule.id, policy, documents, memberships).changed_paths + ) export_membership_map(map_rows, settings.membership_map_path) selected = { str(policy.root / path) for path in paths if isinstance(path, str) } + selected.update(str(policy.root / path) for path in changed) + selected.update(str(policy.root / path) for path in replay_paths) requested_ids = {item for item in collections_requested if isinstance(item, str)} for path, ids in memberships.membership_map_rows(policy.root): if requested_ids.intersection(ids): @@ -144,7 +165,9 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) - app.state.auth = auth_service app.state.engine = engine app.state.path_policy = policy - app.state.collections = CollectionService(SQLiteCollectionRepository(database)) + collection_repository = SQLiteCollectionRepository(database) + app.state.collections = CollectionService(collection_repository) + app.state.collection_repository = collection_repository app.state.memberships = memberships app.state.documents = documents app.state.jobs = jobs diff --git a/src/recoll_modern/domain/membership.py b/src/recoll_modern/domain/membership.py index 52ae764..e77cfec 100644 --- a/src/recoll_modern/domain/membership.py +++ b/src/recoll_modern/domain/membership.py @@ -57,7 +57,7 @@ class MembershipService: } for source, report in discovered: if source.type == "directory" and source.mode == "persistent": - rule = self._memberships.create_rule( + rule = self._memberships.ensure_rule( collection_id, self._policy.normalize(source.path), tuple(self._policy.normalize(path) for path in source.excluded_directories), diff --git a/src/recoll_modern/indexing/reconcile.py b/src/recoll_modern/indexing/reconcile.py index 6351a62..eec0cf3 100644 --- a/src/recoll_modern/indexing/reconcile.py +++ b/src/recoll_modern/indexing/reconcile.py @@ -28,18 +28,31 @@ def reconcile_rule( mode="persistent", excluded_directories=rule.excluded_directories, )]) - skipped = (*rule.excluded_directories, *(error.relative_path for error in report.errors)) + failed_paths = tuple(error.relative_path for error in report.errors) + skipped = (*rule.excluded_directories, *failed_paths) + existing_paths = documents.known_paths_under(rule.relative_path) known_paths = tuple( - path for path in documents.known_paths_under(rule.relative_path) + path for path in existing_paths if not any(path == prefix or path.startswith(prefix + "/") for prefix in skipped) ) + retired_paths = set(known_paths) - set(report.relative_paths) + previous_hashes = memberships.rule_documents(rule_id) + protected_hashes = { + record.content_hash + for path in existing_paths + if any(path == failed or path.startswith(failed + "/") for failed in failed_paths) + if (record := documents.by_path(path)) is not None + and record.content_hash in previous_hashes + } observed = documents.observe_files(list(report.files), reconcile_paths=known_paths) changed = memberships.replace_rule_members( - rule_id, [file.content_hash for file in report.files] + rule_id, [file.content_hash for file in report.files] + sorted(protected_hashes) ) canonical_changes = [ document.canonical_path for document in observed.documents if document.canonical_path is not None ] - return ReconcileResult(report, tuple(sorted(set(changed + canonical_changes)))) + return ReconcileResult( + report, tuple(sorted(set(changed + canonical_changes) | retired_paths)) + ) diff --git a/src/recoll_modern/storage/collections.py b/src/recoll_modern/storage/collections.py index babaff4..a475d75 100644 --- a/src/recoll_modern/storage/collections.py +++ b/src/recoll_modern/storage/collections.py @@ -5,11 +5,14 @@ from __future__ import annotations import secrets import sqlite3 from datetime import UTC, datetime +from sqlite3 import Connection from recoll_modern.domain.collections import CollectionPage, CollectionPaths, normalize_name from recoll_modern.domain.errors import AppError +from recoll_modern.domain.jobs import Job from recoll_modern.domain.models import Collection, CollectionCreate, CollectionPatch, _Unset from recoll_modern.storage.database import Database +from recoll_modern.storage.jobs import JobRepository class SQLiteCollectionRepository: @@ -113,32 +116,47 @@ class SQLiteCollectionRepository: def delete(self, collection_id: str) -> CollectionPaths: """Return effective canonical paths before SQLite cascades memberships away.""" with self._database.transaction() as connection: - paths = [ - str(row[0]) - for row in connection.execute( - "SELECT DISTINCT paths.relative_path " - "FROM document_paths AS paths " - "JOIN (" - " SELECT direct.document_hash " - " FROM direct_memberships AS direct " - " WHERE direct.collection_id = ? " - " UNION " - " SELECT rule_memberships.document_hash " - " FROM rule_memberships " - " JOIN directory_rules AS rules ON rules.id = rule_memberships.rule_id " - " WHERE rules.collection_id = ? " - " AND NOT EXISTS (" - " SELECT 1 FROM membership_exclusions AS exclusions " - " WHERE exclusions.collection_id = rules.collection_id " - " AND exclusions.document_hash = rule_memberships.document_hash" - " )" - ") AS members ON members.document_hash = paths.document_hash " - "WHERE paths.is_canonical = 1 " - "ORDER BY paths.relative_path", - (collection_id, collection_id), - ) - ] - connection.execute("DELETE FROM collections WHERE id = ?", (collection_id,)) + return self._delete_in_transaction(connection, collection_id) + + def delete_with_cleanup(self, collection_id: str, jobs: JobRepository) -> Job: + """Delete metadata and queue its index cleanup in one durable transaction.""" + with self._database.transaction() as connection: + paths = self._delete_in_transaction(connection, collection_id) + return jobs.enqueue_in_transaction( + connection, "collection_cleanup", {"version": 1, "paths": paths}, False + ) + + def _delete_in_transaction( + self, connection: Connection, collection_id: str + ) -> CollectionPaths: + paths = [ + str(row[0]) + for row in connection.execute( + "SELECT DISTINCT paths.relative_path " + "FROM document_paths AS paths " + "JOIN (" + " SELECT direct.document_hash " + " FROM direct_memberships AS direct " + " WHERE direct.collection_id = ? " + " UNION " + " SELECT rule_memberships.document_hash " + " FROM rule_memberships " + " JOIN directory_rules AS rules ON rules.id = rule_memberships.rule_id " + " WHERE rules.collection_id = ? " + " AND NOT EXISTS (" + " SELECT 1 FROM membership_exclusions AS exclusions " + " WHERE exclusions.collection_id = rules.collection_id " + " AND exclusions.document_hash = rule_memberships.document_hash" + " )" + ") AS members ON members.document_hash = paths.document_hash " + "WHERE paths.is_canonical = 1 " + "ORDER BY paths.relative_path", + (collection_id, collection_id), + ) + ] + deleted = connection.execute("DELETE FROM collections WHERE id = ?", (collection_id,)) + if deleted.rowcount != 1: + raise _collection_not_found() return paths diff --git a/src/recoll_modern/storage/jobs.py b/src/recoll_modern/storage/jobs.py index 855190e..8f213d3 100644 --- a/src/recoll_modern/storage/jobs.py +++ b/src/recoll_modern/storage/jobs.py @@ -7,6 +7,7 @@ import secrets import sqlite3 from collections.abc import Mapping from datetime import UTC, datetime +from sqlite3 import Connection from recoll_modern.domain.jobs import Job, JobFileError, JobState from recoll_modern.storage.database import Database @@ -19,25 +20,32 @@ class JobRepository: def enqueue( self, kind: str, request: Mapping[str, object], fail_on_file_error: bool ) -> Job: + with self._database.transaction() as connection: + return self.enqueue_in_transaction(connection, kind, request, fail_on_file_error) + + def enqueue_in_transaction( + self, connection: Connection, kind: str, request: Mapping[str, object], + fail_on_file_error: bool, + ) -> Job: + """Write a job in a caller-owned transaction when state and work must be atomic.""" 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), - ) + 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 + return Job( + job_id, kind, JobState.QUEUED, dict(request), fail_on_file_error, 0, 0 + ) def get(self, job_id: str) -> Job | None: with self._database.connect() as connection: @@ -98,6 +106,26 @@ class JobRepository: ) ] + def record_reindex_paths(self, job_id: str, paths: tuple[str, ...]) -> tuple[str, ...]: + """Write paths ahead of a destructive membership mutation for replay.""" + with self._database.transaction() as connection: + row = connection.execute( + "SELECT request_json FROM jobs WHERE id = ? AND state = 'running'", (job_id,) + ).fetchone() + if row is None: + raise ValueError("job is not running") + request = json.loads(str(row[0])) + prior = request.get("reindex_paths", []) + if not isinstance(prior, list) or not all(isinstance(item, str) for item in prior): + raise ValueError("invalid replay paths") + combined = tuple(sorted(set(prior) | set(paths))) + request["reindex_paths"] = list(combined) + connection.execute( + "UPDATE jobs SET request_json = ? WHERE id = ?", + (json.dumps(request, sort_keys=True), job_id), + ) + return combined + 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: @@ -156,11 +184,49 @@ def _valid_request(kind: str, raw: str) -> bool: return False if not isinstance(value, dict): return False + if value.get("version") != 1: + return False + replay_paths = value.get("reindex_paths", []) + if not _string_list(replay_paths): + return False if kind == "membership_add": + sources = value.get("sources") + if not isinstance(value.get("collection_id"), str) or not isinstance(sources, list): + return False + return all(_valid_source(source) for source in sources) + if kind == "membership_remove": return isinstance(value.get("collection_id"), str) and isinstance( - value.get("sources"), list + value.get("document_hash"), str ) - return bool(kind) + if kind == "rule_patch": + return isinstance(value.get("rule_id"), str) and _string_list( + value.get("excluded_directories") + ) + if kind == "rule_delete": + return isinstance(value.get("rule_id"), str) + if kind == "collection_cleanup": + return _string_list(value.get("paths")) + if kind == "reindex": + return _string_list(value.get("paths")) and _string_list(value.get("collections")) + if kind == "reconcile_rules": + return True + return False + + +def _string_list(value: object) -> bool: + return isinstance(value, list) and all(isinstance(item, str) for item in value) + + +def _valid_source(value: object) -> bool: + if not isinstance(value, dict) or not isinstance(value.get("path"), str): + return False + if value.get("type") == "file": + return True + if value.get("type") == "directory": + return value.get("mode") in ("snapshot", "persistent") and _string_list( + value.get("excluded_directories", []) + ) + return False def _job_from_row(row: sqlite3.Row) -> Job: diff --git a/src/recoll_modern/storage/membership.py b/src/recoll_modern/storage/membership.py index ecc2e67..cba02a6 100644 --- a/src/recoll_modern/storage/membership.py +++ b/src/recoll_modern/storage/membership.py @@ -46,6 +46,26 @@ class MembershipRepository: raise return DirectoryRule(rule_id, collection_id, relative_path, excluded_directories) + def ensure_rule( + self, collection_id: str, relative_path: str, excluded_directories: tuple[str, ...] + ) -> DirectoryRule: + """Replay a persistent source without creating a duplicate rule.""" + with self._database.connect() as connection: + row = connection.execute( + "SELECT id FROM directory_rules WHERE collection_id = ? AND relative_path = ?", + (collection_id, relative_path), + ).fetchone() + if row is None: + return self.create_rule(collection_id, relative_path, excluded_directories) + rule_id = str(row[0]) + rule = self.get_rule(rule_id) + assert rule is not None + if rule.excluded_directories != excluded_directories: + self.patch_rule(rule_id, excluded_directories) + rule = self.get_rule(rule_id) + assert rule is not None + return rule + def get_rule(self, rule_id: str) -> DirectoryRule | None: with self._database.connect() as connection: row = connection.execute( diff --git a/tests/api/test_collections.py b/tests/api/test_collections.py index cd0dd69..1956035 100644 --- a/tests/api/test_collections.py +++ b/tests/api/test_collections.py @@ -1,5 +1,6 @@ from typing import cast +import pytest from fastapi import FastAPI from fastapi.testclient import TestClient @@ -45,4 +46,24 @@ def test_empty_create_and_rename_do_not_enqueue_jobs( client.patch(f"/collections/{collection_id}", json={"name": "Reading"}, headers=headers) with cast(FastAPI, client.app).state.database.connect() as connection: total = connection.execute("SELECT COUNT(*) FROM jobs").fetchone()[0] - assert total == 0 + assert total == 0 + + +def test_delete_rolls_back_when_cleanup_job_cannot_be_written( + authenticated: tuple[TestClient, str], monkeypatch: pytest.MonkeyPatch +) -> None: + client, csrf = authenticated + collection_id = client.post( + "/collections", json={"name": "Books"}, headers={"X-CSRF-Token": csrf} + ).json()["id"] + + def fail_id(_size: int) -> str: + raise RuntimeError("simulated job ID failure") + + monkeypatch.setattr("recoll_modern.storage.jobs.secrets.token_hex", fail_id) + with pytest.raises(RuntimeError, match="simulated job ID failure"): + client.delete( + f"/collections/{collection_id}", params={"confirmation": "Books"}, + headers={"X-CSRF-Token": csrf}, + ) + assert client.get(f"/collections/{collection_id}").status_code == 200 diff --git a/tests/api/test_documents.py b/tests/api/test_documents.py index d6858dc..b9a16fe 100644 --- a/tests/api/test_documents.py +++ b/tests/api/test_documents.py @@ -34,6 +34,8 @@ def test_text_and_byte_range(authenticated: tuple[TestClient, str], tmp_path: Pa assert client.get( f"/documents/{digest}/file", headers={"Range": "bytes=99-100"} ).status_code == 416 + suffix = client.get(f"/documents/{digest}/file", headers={"Range": "bytes=-2"}) + assert suffix.status_code == 206 and suffix.content == b"ef" def test_unknown_and_missing_documents( diff --git a/tests/api/test_jobs.py b/tests/api/test_jobs.py index 2a8bc0b..2b3ff97 100644 --- a/tests/api/test_jobs.py +++ b/tests/api/test_jobs.py @@ -1,5 +1,8 @@ import time +from pathlib import Path +from typing import cast +from fastapi import FastAPI from fastapi.testclient import TestClient @@ -22,6 +25,7 @@ def test_retry_refuses_completed_job(authenticated: tuple[TestClient, str]) -> N if client.get(f"/jobs/{job_id}").json()["job"]["state"] == "completed": break time.sleep(0.01) + time.sleep(0.01) response = client.post(f"/jobs/{job_id}/retry", headers={"X-CSRF-Token": csrf}) assert response.status_code == 409 assert response.json()["error"]["code"] == "job_not_retryable" @@ -32,3 +36,37 @@ def test_job_list_validates_pagination(authenticated: tuple[TestClient, str]) -> response = client.get("/jobs", params={"limit": 201}) assert response.status_code == 422 assert response.json()["error"]["code"] == "validation_failed" + + +def test_collection_reindex_includes_departed_rule_path( + authenticated: tuple[TestClient, str], tmp_path: Path +) -> None: + client, csrf = authenticated + books = tmp_path / "docs" / "books" + books.mkdir() + file = books / "gone.txt" + file.write_text("gone") + headers = {"X-CSRF-Token": csrf} + collection_id = client.post( + "/collections", headers=headers, json={"name": "Books"} + ).json()["id"] + created = client.post( + f"/collections/{collection_id}/memberships", headers=headers, + json={"sources": [{"type": "directory", "path": "books", "mode": "persistent"}]}, + ) + job_id = created.json()["job"]["id"] + for _ in range(100): + if client.get(f"/jobs/{job_id}").json()["job"]["state"] == "completed": + break + engine = cast(FastAPI, client.app).state.engine + engine.forced_paths.clear() + file.unlink() + refreshed = client.post( + "/jobs/reindex", headers=headers, json={"collections": [collection_id]} + ) + refresh_id = refreshed.json()["job"]["id"] + for _ in range(100): + if client.get(f"/jobs/{refresh_id}").json()["job"]["state"] == "completed": + break + time.sleep(0.01) + assert any(str(file) in batch for batch in engine.forced_paths) diff --git a/tests/api/test_memberships.py b/tests/api/test_memberships.py index 53398e0..6720b9b 100644 --- a/tests/api/test_memberships.py +++ b/tests/api/test_memberships.py @@ -2,6 +2,8 @@ import time from pathlib import Path from typing import cast +import pytest +from fastapi import FastAPI from fastapi.testclient import TestClient @@ -177,3 +179,90 @@ def test_remove_document_creates_override_then_direct_add_restores_it( assert len(members) == 1 assert members[0]["direct"] is True assert members[0]["override"] is False + + +def test_rule_deletion_retry_reindexes_removed_membership( + authenticated: tuple[TestClient, str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + client, csrf = authenticated + books = tmp_path / "docs" / "books" + books.mkdir() + file = books / "a.txt" + file.write_text("book") + collection_id = _collection(client, csrf) + created = client.post( + f"/collections/{collection_id}/memberships", + headers={"X-CSRF-Token": csrf}, + json={"sources": [{"type": "directory", "path": "books", "mode": "persistent"}]}, + ) + assert _wait(client, created.json()["job"]["id"])["state"] == "completed" + rule_id = client.get(f"/collections/{collection_id}/directory-rules").json()["items"][0]["id"] + engine = cast(FastAPI, client.app).state.engine + original = engine.force_reindex + calls: list[tuple[str, ...]] = [] + + def fail_once(paths: tuple[str, ...]) -> None: + calls.append(paths) + if len(calls) == 1: + raise RuntimeError("simulated indexing failure") + original(paths) + + monkeypatch.setattr(engine, "force_reindex", fail_once) + deleted = client.delete( + f"/collections/{collection_id}/directory-rules/{rule_id}", + headers={"X-CSRF-Token": csrf}, + ) + job_id = deleted.json()["job"]["id"] + assert _wait(client, job_id)["state"] == "failed" + retried = client.post(f"/jobs/{job_id}/retry", headers={"X-CSRF-Token": csrf}) + assert retried.status_code == 202 + assert _wait(client, job_id)["state"] == "completed" + assert calls == [(str(file),), (str(file),)] + + +def test_rule_patch_retry_reindexes_excluded_member( + authenticated: tuple[TestClient, str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + client, csrf = authenticated + skipped = tmp_path / "docs" / "books" / "skip" + skipped.mkdir(parents=True) + file = skipped / "a.txt" + file.write_text("book") + collection_id = _collection(client, csrf) + created = client.post( + f"/collections/{collection_id}/memberships", + headers={"X-CSRF-Token": csrf}, + json={"sources": [{"type": "directory", "path": "books", "mode": "persistent"}]}, + ) + assert _wait(client, created.json()["job"]["id"])["state"] == "completed" + rule_id = client.get(f"/collections/{collection_id}/directory-rules").json()["items"][0]["id"] + engine = cast(FastAPI, client.app).state.engine + original = engine.force_reindex + calls: list[tuple[str, ...]] = [] + + def fail_once(paths: tuple[str, ...]) -> None: + calls.append(paths) + if len(calls) == 1: + raise RuntimeError("simulated indexing failure") + original(paths) + + monkeypatch.setattr(engine, "force_reindex", fail_once) + patched = client.patch( + f"/collections/{collection_id}/directory-rules/{rule_id}", + headers={"X-CSRF-Token": csrf}, json={"excluded_directories": ["books/skip"]}, + ) + job_id = patched.json()["job"]["id"] + assert _wait(client, job_id)["state"] == "failed" + client.post(f"/jobs/{job_id}/retry", headers={"X-CSRF-Token": csrf}) + assert _wait(client, job_id)["state"] == "completed" + assert calls == [(str(file),), (str(file),)] + + +def test_rule_list_validates_pagination(authenticated: tuple[TestClient, str]) -> None: + client, csrf = authenticated + collection_id = _collection(client, csrf) + response = client.get( + f"/collections/{collection_id}/directory-rules", params={"limit": 201} + ) + assert response.status_code == 422 + assert response.json()["error"]["code"] == "validation_failed" diff --git a/tests/contract/test_recoll_engine.py b/tests/contract/test_recoll_engine.py index 9636e44..722189f 100644 --- a/tests/contract/test_recoll_engine.py +++ b/tests/contract/test_recoll_engine.py @@ -74,13 +74,10 @@ class TestRecollEngine(EngineContract): if config is None: if os.environ.get("RECOLL_MODERN_REQUIRE_DOCKER") != "1": pytest.skip("Recoll test container is not running") - corpus_file = os.environ.get("RECOLL_MODERN_TEST_CORPUS_FILE") - if corpus_file is None: - pytest.fail("RECOLL_MODERN_TEST_CORPUS_FILE is required for container contract") bootstrap = subprocess.run( ["docker", "compose", "exec", "-T", "app", "recollindex", "-c", "/var/lib/recoll-modern/recoll-config", "-i", "-Z", - f"/documents/{corpus_file}"], + "/documents"], capture_output=True, text=True, check=False, timeout=30, ) assert bootstrap.returncode == 0, bootstrap.stderr diff --git a/tests/indexing/test_job_handlers.py b/tests/indexing/test_job_handlers.py index 2c53c88..d8d2f91 100644 --- a/tests/indexing/test_job_handlers.py +++ b/tests/indexing/test_job_handlers.py @@ -89,3 +89,36 @@ def test_handler_exports_map_before_forced_reindex(tmp_path: Path) -> None: handler.handle(claimed) assert calls == [(str(root / "good.txt"),)] assert jobs.get(job.id).state == JobState.COMPLETED # type: ignore[union-attr] + + +def test_persistent_source_replays_after_index_failure(tmp_path: Path) -> None: + root, jobs, membership, _, collection_id = _setup(tmp_path) + (root / "books").mkdir() + (root / "books" / "a.txt").write_text("book") + forced: list[tuple[str, ...]] = [] + + class FlakyIndexer: + def force_reindex(self, paths: tuple[str, ...]) -> None: + forced.append(paths) + if len(forced) == 1: + raise RuntimeError("simulated index failure") + + handler = MembershipJobHandler( + PathPolicy(root), DocumentRepository(Database(tmp_path / "state.sqlite3")), + membership, jobs, tmp_path / "membership.map", FlakyIndexer(), + ) + job = jobs.enqueue("membership_add", { + "version": 1, "collection_id": collection_id, + "sources": [{"type": "directory", "path": "books", "mode": "persistent", + "excluded_directories": []}], + }, False) + claimed = jobs.claim_next() + assert claimed is not None + try: + handler.handle(claimed) + except RuntimeError: + pass + assert len(membership.list_rules()) == 1 + handler.handle(claimed) + assert jobs.get(job.id).state == JobState.COMPLETED # type: ignore[union-attr] + assert forced == [(str(root / "books" / "a.txt"),)] * 2 diff --git a/tests/indexing/test_reconcile.py b/tests/indexing/test_reconcile.py index 115d26a..74bab7c 100644 --- a/tests/indexing/test_reconcile.py +++ b/tests/indexing/test_reconcile.py @@ -1,6 +1,9 @@ from pathlib import Path +import pytest + from recoll_modern.domain.models import CollectionCreate +from recoll_modern.indexing.discovery import DiscoveryReport, FileError from recoll_modern.indexing.paths import PathPolicy from recoll_modern.indexing.reconcile import reconcile_rule from recoll_modern.storage.collections import SQLiteCollectionRepository @@ -52,3 +55,30 @@ def test_excluded_file_is_not_marked_missing(tmp_path: Path) -> None: record = documents.get(content_hash) assert record is not None and record.state == "present" assert membership.rule_documents(rule.id) == set() + + +def test_unreadable_file_retains_rule_membership( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "docs" + books = root / "books" + books.mkdir(parents=True) + (books / "a.txt").write_text("a") + database = Database(tmp_path / "state.sqlite3") + database.migrate() + collection_id = SQLiteCollectionRepository(database).create(CollectionCreate("Books")).id + membership = MembershipRepository(database) + documents = DocumentRepository(database) + rule = membership.create_rule(collection_id, "books", ()) + first = reconcile_rule(rule.id, PathPolicy(root), documents, membership) + content_hash = first.discovery.files[0].content_hash + monkeypatch.setattr( + "recoll_modern.indexing.reconcile.discover_sources", + lambda _policy, _sources: DiscoveryReport( + (), (FileError("books/a.txt", "file_unreadable"),) + ), + ) + second = reconcile_rule(rule.id, PathPolicy(root), documents, membership) + assert second.discovery.errors[0].code == "file_unreadable" + assert membership.rule_documents(rule.id) == {content_hash} + assert documents.get(content_hash).state == "present" # type: ignore[union-attr] diff --git a/tests/indexing/test_worker.py b/tests/indexing/test_worker.py index 60d3830..8504ebd 100644 --- a/tests/indexing/test_worker.py +++ b/tests/indexing/test_worker.py @@ -48,10 +48,10 @@ 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) + job = jobs.enqueue("reindex", {"version": 1, "paths": [], "collections": []}, False) jobs.claim_next() handled: list[str] = [] - worker = JobWorker(jobs, {"test": lambda claimed: handled.append(claimed.id)}) + worker = JobWorker(jobs, {"reindex": lambda claimed: handled.append(claimed.id)}) worker.start() for _ in range(100): current = jobs.get(job.id) diff --git a/tests/integration/test_end_to_end.py b/tests/integration/test_end_to_end.py index 7b448ca..51eba29 100644 --- a/tests/integration/test_end_to_end.py +++ b/tests/integration/test_end_to_end.py @@ -1,9 +1,14 @@ """Exercise the complete HTTP lifecycle with durable state and a deterministic engine.""" +import os +import subprocess import time +import uuid +from collections.abc import Callable from pathlib import Path from typing import cast +import httpx import pytest from fastapi import FastAPI from fastapi.testclient import TestClient @@ -15,6 +20,24 @@ from recoll_modern.storage.database import Database from recoll_modern.storage.jobs import JobRepository +def _live_wait( + client: httpx.Client, path: str, job_id: str, headers: dict[str, str] +) -> dict[str, object]: + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + try: + response = client.get(f"{path}/jobs/{job_id}", headers=headers) + except httpx.TransportError: + time.sleep(0.1) + continue + if response.status_code == 200: + job = cast(dict[str, object], response.json()["job"]) + if job["state"] not in ("queued", "running"): + return job + time.sleep(0.1) + raise AssertionError(f"container job {job_id} did not finish") + + def _wait(client: TestClient, job_id: str) -> dict[str, object]: deadline = time.monotonic() + 5 while time.monotonic() < deadline: @@ -118,3 +141,125 @@ def test_complete_collection_lifecycle(tmp_path: Path, monkeypatch: pytest.Monke "username": "admin", "password": "secret-password", }) assert _wait(client, interrupted.id)["state"] == "completed" + + +def test_real_container_collection_lifecycle( + compose_exec: Callable[[list[str]], subprocess.CompletedProcess[str]], +) -> None: + document_root = os.environ.get("RECOLL_MODERN_DOCUMENTS") + if document_root is None: + pytest.fail("RECOLL_MODERN_DOCUMENTS is required for real-container acceptance") + relative = Path(f"recoll-modern-test-{uuid.uuid4().hex}") + corpus = Path(document_root) / relative + (corpus / "a" / ".calibre").mkdir(parents=True) + (corpus / "b").mkdir() + marker = uuid.uuid4().hex + content = { + "a/shared.txt": f"zebraplatypus duplicate {marker}", + "b/shared.txt": f"zebraplatypus duplicate {marker}", + "a/other.txt": f"zebraplatypus first distinct {marker}", + "b/other.txt": f"zebraplatypus second distinct {marker}", + "a/special.txt": f"zebraplatypus second only {marker}", + "a/.calibre/hidden.txt": f"zebraplatypus excluded {marker}", + } + for name, text in content.items(): + file = corpus / name + file.write_text(text) + prefix = os.environ.get("RECOLL_MODERN_ROOT_PATH", "/").rstrip("/") + port = os.environ.get("RECOLL_MODERN_PUBLISH_PORT", "8000") + with httpx.Client(base_url=f"http://127.0.0.1:{port}", timeout=15) as client: + login = client.post(f"{prefix}/auth/login", json={ + "username": os.environ["RECOLL_MODERN_ADMIN_USERNAME"], + "password": os.environ["RECOLL_MODERN_ADMIN_PASSWORD"], + }) + assert login.status_code == 200, login.text + cookie = login.cookies.get("recoll_modern_session") + assert cookie + headers = { + "Cookie": f"recoll_modern_session={cookie}", + "X-CSRF-Token": login.json()["csrf_token"], + } + a = client.post( + f"{prefix}/collections", headers=headers, + json={"name": f"Real A {uuid.uuid4().hex}"}, + ).json()["id"] + b = client.post( + f"{prefix}/collections", headers=headers, + json={"name": f"Real B {uuid.uuid4().hex}"}, + ).json()["id"] + def add(collection_id: str, sources: list[dict[str, object]]) -> str: + response = client.post( + f"{prefix}/collections/{collection_id}/memberships", headers=headers, + json={"sources": sources}, + ) + assert response.status_code == 202, response.text + job_id = response.json()["job"]["id"] + assert _live_wait(client, prefix, job_id, headers)["state"] == "completed" + return cast(str, job_id) + + rule_job = add(a, [{ + "type": "directory", "path": (relative / "a").as_posix(), + "mode": "persistent", + "excluded_directories": [(relative / "a" / ".calibre").as_posix()], + }]) + add(b, [{"type": "directory", "path": (relative / "b").as_posix(), + "mode": "snapshot"}]) + add(b, [{"type": "file", "path": (relative / "a" / "special.txt").as_posix()}]) + docs_a = client.get(f"{prefix}/collections/{a}/documents", headers=headers).json()["items"] + docs_b = client.get(f"{prefix}/collections/{b}/documents", headers=headers).json()["items"] + assert len(docs_a) == 3 and len(docs_b) == 3 + assert all(".calibre" not in item["path"] for item in docs_a) + shared = next( + item for item in docs_a + if item["document_id"] in {other["document_id"] for other in docs_b} + and item["path"].endswith("shared.txt") + ) + assert shared["document_id"] in {item["document_id"] for item in docs_b} + combined = client.get(f"{prefix}/search", params=[ + ("q", "zebraplatypus"), ("collections", a), ("collections", b), + ("fields", "text"), ("text_limit", "8"), + ], headers=headers) + assert combined.status_code == 200, combined.text + assert len(combined.json()["hits"]) == 4, combined.json() + assert all(hit["text"] == "zebrapla" for hit in combined.json()["hits"]) + text_response = client.get( + f"{prefix}/documents/{shared['document_id']}/text", headers=headers + ) + assert text_response.status_code == 200 and "duplicate" in text_response.json()["text"] + partial = client.get( + f"{prefix}/documents/{shared['document_id']}/file", + headers={**headers, "Range": "bytes=0-7"}, + ) + assert partial.status_code == 206 and partial.content == b"zebrapla" + + (corpus / "a" / "shared.txt").unlink() + refreshed = client.post(f"{prefix}/jobs/reindex", headers=headers, json={}) + assert _live_wait(client, prefix, refreshed.json()["job"]["id"], headers)[ + "state" + ] == "completed" + promoted = client.get(f"{prefix}/collections/{b}/documents", headers=headers).json()[ + "items" + ] + assert next(item for item in promoted if item["document_id"] == shared["document_id"])[ + "path" + ] == (relative / "b" / "shared.txt").as_posix() + assert client.get(f"{prefix}/search", params={ + "q": "zebraplatypus", "collections": b, + }, headers=headers).json()["hits"] + + marked = compose_exec(["python", "-c", ( + "from recoll_modern.storage.database import Database; " + "from pathlib import Path; " + "db=Database(Path('/var/lib/recoll-modern/recoll-modern.sqlite3')); " + "c=db.connect(); " + "c.execute(\"UPDATE jobs SET state='running' WHERE id=?\", " + "(__import__('sys').argv[1],)); " + "c.commit(); c.close()" + ), rule_job]) + assert marked.returncode == 0, marked.stderr + restarted = subprocess.run( + ["docker", "compose", "restart", "app"], capture_output=True, text=True, + check=False, timeout=40, + ) + assert restarted.returncode == 0, restarted.stderr + assert _live_wait(client, prefix, rule_job, headers)["state"] == "completed" diff --git a/tests/integration/test_read_only_corpus.py b/tests/integration/test_read_only_corpus.py index a996fce..061b2db 100644 --- a/tests/integration/test_read_only_corpus.py +++ b/tests/integration/test_read_only_corpus.py @@ -3,7 +3,9 @@ import os import subprocess import time +import uuid from collections.abc import Callable +from pathlib import Path import httpx import pytest @@ -21,7 +23,14 @@ def test_generated_corpus_indexes_and_searches( ) -> None: corpus_file = os.environ.get("RECOLL_MODERN_TEST_CORPUS_FILE") if corpus_file is None: - pytest.skip("Set RECOLL_MODERN_TEST_CORPUS_FILE for generated-corpus verification") + document_root = os.environ.get("RECOLL_MODERN_DOCUMENTS") + if document_root is None: + pytest.fail("RECOLL_MODERN_DOCUMENTS is required for generated-corpus verification") + relative = Path(f"recoll-modern-test-{uuid.uuid4().hex}") / "probe.txt" + file = Path(document_root) / relative + file.parent.mkdir(parents=True) + file.write_text("Unique zebraplatypus integration corpus.\n") + corpus_file = relative.as_posix() assert compose_exec(["test", "-f", f"/documents/{corpus_file}"]).returncode == 0 port = os.environ.get("RECOLL_MODERN_PUBLISH_PORT", "8000") username = os.environ["RECOLL_MODERN_ADMIN_USERNAME"] diff --git a/tests/security/test_downloads.py b/tests/security/test_downloads.py index 9ec1e6d..7eda26d 100644 --- a/tests/security/test_downloads.py +++ b/tests/security/test_downloads.py @@ -6,6 +6,7 @@ from typing import cast from fastapi import FastAPI from fastapi.testclient import TestClient +from recoll_modern.api.documents import open_contained_file from recoll_modern.indexing.discovery import ObservedFile @@ -23,3 +24,19 @@ def test_missing_document_is_gone( response = client.get(f"/documents/{digest}/file") assert response.status_code == 410 assert response.json()["error"]["code"] == "document_missing" + + +def test_open_descriptor_cannot_be_redirected_by_later_symlink_swap(tmp_path: Path) -> None: + root = tmp_path / "docs" + root.mkdir() + file = root / "safe.txt" + file.write_text("safe") + outside = tmp_path / "secret.txt" + outside.write_text("secret") + opened = open_contained_file(root, "safe.txt") + file.unlink() + file.symlink_to(outside) + try: + assert opened.read() == b"safe" + finally: + opened.close() diff --git a/tests/storage/test_job_repository.py b/tests/storage/test_job_repository.py index 6ffc90e..2769410 100644 --- a/tests/storage/test_job_repository.py +++ b/tests/storage/test_job_repository.py @@ -1,5 +1,7 @@ from pathlib import Path +import pytest + from recoll_modern.domain.jobs import JobFileError, JobState from recoll_modern.storage.database import Database from recoll_modern.storage.jobs import JobRepository @@ -11,10 +13,14 @@ def _repo(tmp_path: Path) -> JobRepository: return JobRepository(database) +def _membership_request(sources: object = None) -> dict[str, object]: + return {"version": 1, "collection_id": "c1", "sources": [] if sources is None else sources} + + 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) + first = repo.enqueue("membership_add", _membership_request(), False) + second = repo.enqueue("membership_add", _membership_request(), 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 @@ -22,7 +28,7 @@ def test_claim_next_is_durable_and_oldest_first(tmp_path: Path) -> 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) + job = repo.enqueue("membership_add", _membership_request(), False) repo.claim_next() repo.recover_running() assert repo.get(job.id).state == JobState.QUEUED # type: ignore[union-attr] @@ -32,7 +38,7 @@ 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) + job = repo.enqueue("membership_add", _membership_request(), False) repo.claim_next() with database.transaction() as connection: connection.execute("UPDATE jobs SET request_json = ? WHERE id = ?", ("{", job.id)) @@ -46,7 +52,7 @@ 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) + job = repo.enqueue("membership_add", _membership_request("not a list"), False) repo.claim_next() repo.recover_running() assert repo.get(job.id).state == JobState.FAILED # type: ignore[union-attr] @@ -54,7 +60,7 @@ def test_recover_running_fails_invalid_source_shape(tmp_path: Path) -> None: 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) + job = repo.enqueue("membership_add", _membership_request(), False) repo.claim_next() repo.add_file_error(job.id, JobFileError("gone.txt", "file_vanished", "File vanished")) repo.recover_running() @@ -63,8 +69,27 @@ def test_recover_running_clears_errors_from_interrupted_attempt(tmp_path: Path) 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) + job = repo.enqueue("membership_add", _membership_request(), 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] + + +@pytest.mark.parametrize("kind,payload", [ + ("rule_delete", {"version": 1}), + ("rule_patch", {"version": 1, "rule_id": "r1", "excluded_directories": "bad"}), + ("reindex", {"version": 1, "paths": "bad", "collections": []}), + ("collection_cleanup", {"version": 2, "paths": []}), + ("unknown", {"version": 1}), +]) +def test_recovery_rejects_invalid_kind_payload( + tmp_path: Path, kind: str, payload: dict[str, object] +) -> None: + repo = _repo(tmp_path) + job = repo.enqueue(kind, payload, False) + repo.claim_next() + 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"