From f6f03b9128f6ba88c269628d92474e5d4cac1e10 Mon Sep 17 00:00:00 2001 From: Lucas Winther Date: Sat, 26 Sep 2026 10:51:40 +0200 Subject: [PATCH] feat: export Recoll membership metadata --- .gitignore | 1 + docs/ARCHITECTURE.md | 12 ++-- docs/TODO.md | 9 ++- docs/plans/001-search-engine-api.md | 10 +-- helper/Makefile | 12 ++++ helper/recoll_membership.c | 90 ++++++++++++++++++++++++ src/recoll_modern/indexing/handlers.py | 21 +++++- src/recoll_modern/indexing/map_export.py | 68 ++++++++++++++++++ src/recoll_modern/storage/membership.py | 14 ++++ tests/helper/test_membership_helper.py | 69 ++++++++++++++++++ tests/indexing/test_job_handlers.py | 35 ++++++++- tests/indexing/test_map_export.py | 66 +++++++++++++++++ 12 files changed, 391 insertions(+), 16 deletions(-) create mode 100644 helper/Makefile create mode 100644 helper/recoll_membership.c create mode 100644 src/recoll_modern/indexing/map_export.py create mode 100644 tests/helper/test_membership_helper.py create mode 100644 tests/indexing/test_map_export.py diff --git a/.gitignore b/.gitignore index f02d974..d19192f 100644 --- a/.gitignore +++ b/.gitignore @@ -30,6 +30,7 @@ htmlcov/ *.sqlite3-shm *.sqlite3-wal membership.map +helper/recoll-membership # Recoll runtime data (configuration templates remain tracked) recoll.conf diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 457dba3..89c1fee 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -26,6 +26,8 @@ web UI and (b) an MCP server, running in Docker, with several independent collec | `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 | +| `src/recoll_modern/indexing/map_export.py` | Writes a durable atomic map from canonical paths to opaque collection IDs | +| `helper/` | C reader run by Recoll's `metadatacmds` for each indexed file | | `tests/` | Automated service tests | Discovery produces content hashes and root-relative paths without writing to the document volume. @@ -69,10 +71,12 @@ future distribution track and starts only after the hosted product through 004 i | `engine/` | The search/index Protocol, fake implementation and Recoll adapter | | `app.py` | Configuration, dependency wiring and lifecycle | -Only `engine/` may import Recoll or invoke `recollindex`. Membership changes first establish desired -state in SQLite, atomically export the helper's flat map, then force-index affected canonical paths. -Engine retries are therefore idempotent. Persistent directory rules reconcile at startup and on an -explicit reindex, not through live filesystem monitoring. +Only `engine/` may import Recoll or invoke `recollindex`. Membership jobs first establish desired +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, +not through live filesystem monitoring. ### Storage model diff --git a/docs/TODO.md b/docs/TODO.md index e1ebfcc..e5484e6 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -15,12 +15,11 @@ 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–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. +- Plan Tasks 1–3 are complete and independently reviewed. Tasks 4–7 add discovery, content-hash + identity, additive membership, durable jobs, and the atomic membership map with its C helper. Their + checkboxes are current in the plan. The next task is the engine contract and Recoll adapter. -The next action is Task 7, "Export membership atomically and build the native helper." Do not repeat -Tasks 1–6. +The next action is Task 8, "Define the engine contract and Recoll adapter." Do not repeat Tasks 1–7. 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 6242dfd..8644c70 100644 --- a/docs/plans/001-search-engine-api.md +++ b/docs/plans/001-search-engine-api.md @@ -530,7 +530,7 @@ git commit -m "feat: add durable indexing jobs" - Produces: `export_membership_map(repository, destination: Path) -> None` - Produces: `recoll-membership FILE_PATH`, reading `RECOLL_MODERN_MEMBERSHIP_MAP` -- [ ] **Step 1: Write failing binary-format and helper tests** +- [x] **Step 1: Write failing binary-format and helper tests** ```python def test_export_is_atomic_and_handles_newlines(tmp_path: Path) -> None: @@ -546,21 +546,21 @@ def test_helper_returns_sorted_space_separated_ids(compiled_helper: Path, map_fi assert result.stdout == "c0000000001 c0000000002\n" ``` -- [ ] **Step 2: Run tests and observe failure** +- [x] **Step 2: Run tests and observe failure** Run: `uv run pytest tests/indexing/test_map_export.py tests/helper/test_membership_helper.py -v` -- [ ] **Step 3: Implement the robust flat-map format and C reader** +- [x] **Step 3: Implement the robust flat-map format and C reader** The format is `RCM1`, followed by sorted records of big-endian `uint32 path_length`, raw UTF-8 path bytes, big-endian `uint32 value_length`, and ASCII space-separated collection IDs. The exporter writes and fsyncs a sibling temporary file, chmods it `0640`, `os.replace()`s it, then fsyncs the parent directory. Paths in this private map are absolute because Recoll passes `%f`; no API exposes them. The C helper opens the configured map read-only, validates every length against file size and a 16 MiB ceiling, scans records with `memcmp`, writes only the matched value plus newline, returns 0 for no membership, and returns nonzero with a generic stderr message for corrupt/unreadable maps. Compile with `cc -std=c17 -O2 -Wall -Wextra -Werror`. -- [ ] **Step 4: Wire export before every forced reindex** +- [x] **Step 4: Wire export before every forced reindex** Extend `MembershipJobHandler` so a successful desired-state transaction calls `export_membership_map` once, then invokes the engine with the distinct sorted changed canonical paths. Add a failure-injection test proving the old map remains intact if write or replace fails. -- [ ] **Step 5: Verify and commit** +- [x] **Step 5: Verify and commit** Run: `make -C helper clean all && uv run pytest tests/indexing/test_map_export.py tests/helper/test_membership_helper.py tests/indexing/test_job_handlers.py -v && uv run ruff check . && uv run mypy src tests` diff --git a/helper/Makefile b/helper/Makefile new file mode 100644 index 0000000..5673bd7 --- /dev/null +++ b/helper/Makefile @@ -0,0 +1,12 @@ +CC ?= cc +CFLAGS ?= -std=c17 -O2 -Wall -Wextra -Werror + +all: recoll-membership + +recoll-membership: recoll_membership.c + $(CC) $(CFLAGS) $< -o $@ + +clean: + rm -f recoll-membership + +.PHONY: all clean diff --git a/helper/recoll_membership.c b/helper/recoll_membership.c new file mode 100644 index 0000000..a49cfd8 --- /dev/null +++ b/helper/recoll_membership.c @@ -0,0 +1,90 @@ +#define _POSIX_C_SOURCE 200809L + +#include +#include +#include +#include +#include +#include + +#define MAX_FIELD (16u * 1024u * 1024u) + +static uint32_t read_u32(const unsigned char *bytes) { + return ((uint32_t)bytes[0] << 24) | ((uint32_t)bytes[1] << 16) | + ((uint32_t)bytes[2] << 8) | (uint32_t)bytes[3]; +} + +static int error(const char *message) { + fprintf(stderr, "recoll-membership: %s\n", message); + return 1; +} + +int main(int argc, char **argv) { + const char *map_path = getenv("RECOLL_MODERN_MEMBERSHIP_MAP"); + if (argc != 2 || map_path == NULL || map_path[0] == '\0') { + return error("map or file path is missing"); + } + FILE *stream = fopen(map_path, "rb"); + if (stream == NULL) { + return error("map is unreadable"); + } + struct stat info; + if (fstat(fileno(stream), &info) != 0 || info.st_size < 4 || + (uintmax_t)info.st_size > SIZE_MAX) { + fclose(stream); + return error("map is corrupt"); + } + size_t size = (size_t)info.st_size; + unsigned char *data = malloc(size); + if (data == NULL || fread(data, 1, size, stream) != size) { + free(data); + fclose(stream); + return error("map is unreadable"); + } + fclose(stream); + if (memcmp(data, "RCM1", 4) != 0) { + free(data); + return error("map is corrupt"); + } + const unsigned char *match = NULL; + size_t match_length = 0; + size_t path_length = strlen(argv[1]); + size_t offset = 4; + while (offset < size) { + if (size - offset < 4) { + free(data); + return error("map is corrupt"); + } + size_t key_length = read_u32(data + offset); + offset += 4; + if (key_length > MAX_FIELD || key_length > size - offset) { + free(data); + return error("map is corrupt"); + } + const unsigned char *key = data + offset; + offset += key_length; + if (size - offset < 4) { + free(data); + return error("map is corrupt"); + } + size_t value_length = read_u32(data + offset); + offset += 4; + if (value_length > MAX_FIELD || value_length > size - offset) { + free(data); + return error("map is corrupt"); + } + if (key_length == path_length && memcmp(key, argv[1], path_length) == 0) { + match = data + offset; + match_length = value_length; + } + offset += value_length; + } + if (match != NULL && match_length != 0) { + if (fwrite(match, 1, match_length, stdout) != match_length || putchar('\n') == EOF) { + free(data); + return error("output failed"); + } + } + free(data); + return 0; +} diff --git a/src/recoll_modern/indexing/handlers.py b/src/recoll_modern/indexing/handlers.py index b8d9b0c..e70428a 100644 --- a/src/recoll_modern/indexing/handlers.py +++ b/src/recoll_modern/indexing/handlers.py @@ -1,14 +1,22 @@ """Durable membership job application after complete source discovery.""" +from pathlib import Path +from typing import Protocol + 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.map_export import MembershipMapRows, export_membership_map 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 ForceIndexer(Protocol): + def force_reindex(self, paths: tuple[str, ...]) -> None: ... + + class MembershipJobHandler: def __init__( self, @@ -16,9 +24,14 @@ class MembershipJobHandler: documents: DocumentRepository, memberships: MembershipRepository, jobs: JobRepository, + membership_map_path: Path, + indexer: ForceIndexer, ) -> None: self._policy = policy self._service = MembershipService(policy, documents, memberships) + self._map_rows = MembershipMapRows(memberships, policy.root) + self._map_path = membership_map_path + self._indexer = indexer self._jobs = jobs def handle(self, job: Job) -> None: @@ -43,10 +56,16 @@ class MembershipJobHandler: self._jobs.fail(job.id, "file_error", "One or more files could not be read") return try: - self._service.apply_discovered(collection_id, discovered) + result = self._service.apply_discovered(collection_id, discovered) except ValueError: self._jobs.fail(job.id, "invalid_job_request", "Job request is invalid") return + export_membership_map(self._map_rows, self._map_path) + changed_paths = tuple( + str(self._policy.root / path) for path in sorted(set(result.changed_paths)) + ) + if changed_paths: + self._indexer.force_reindex(changed_paths) self._jobs.progress(job.id, total, total) self._jobs.complete(job.id, with_errors=bool(errors)) diff --git a/src/recoll_modern/indexing/map_export.py b/src/recoll_modern/indexing/map_export.py new file mode 100644 index 0000000..54e8afe --- /dev/null +++ b/src/recoll_modern/indexing/map_export.py @@ -0,0 +1,68 @@ +"""Atomic binary export of canonical-path collection membership.""" + +import os +import re +import struct +import tempfile +from collections.abc import Iterable, Sequence +from pathlib import Path +from typing import Protocol + +from recoll_modern.storage.membership import MembershipRepository + +_COLLECTION_ID = re.compile(r"c[0-9a-f]{10}\Z") +_MAX_FIELD = 16 * 1024 * 1024 + + +class MembershipRows(Protocol): + def membership_map_rows(self) -> Iterable[tuple[str, Sequence[str]]]: ... + + +class MembershipMapRows: + """Bind a repository's root-relative canonical paths to its document root.""" + + def __init__(self, repository: MembershipRepository, document_root: Path) -> None: + self._repository = repository + self._document_root = document_root + + def membership_map_rows(self) -> Iterable[tuple[str, Sequence[str]]]: + return self._repository.membership_map_rows(self._document_root) + + +def export_membership_map(repository: MembershipRows, destination: Path) -> None: + """Replace a complete map only after its bytes have reached the filesystem.""" + rows = sorted(repository.membership_map_rows(), key=lambda row: row[0]) + temporary: Path | None = None + try: + with tempfile.NamedTemporaryFile( + mode="wb", dir=destination.parent, suffix=".tmp", delete=False + ) as stream: + temporary = Path(stream.name) + stream.write(b"RCM1") + for path, collection_ids in rows: + if not path.startswith("/"): + raise ValueError("membership map paths must be absolute") + ids = sorted(set(collection_ids)) + if any(_COLLECTION_ID.fullmatch(item) is None for item in ids): + raise ValueError("membership map contains an invalid collection ID") + path_bytes = path.encode("utf-8") + value_bytes = " ".join(ids).encode("ascii") + if len(path_bytes) > _MAX_FIELD or len(value_bytes) > _MAX_FIELD: + raise ValueError("membership map field exceeds 16 MiB") + stream.write(struct.pack(">I", len(path_bytes))) + stream.write(path_bytes) + stream.write(struct.pack(">I", len(value_bytes))) + stream.write(value_bytes) + stream.flush() + os.fchmod(stream.fileno(), 0o640) + os.fsync(stream.fileno()) + os.replace(temporary, destination) + temporary = None + directory_fd = os.open(destination.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) diff --git a/src/recoll_modern/storage/membership.py b/src/recoll_modern/storage/membership.py index 0d7ea06..1716029 100644 --- a/src/recoll_modern/storage/membership.py +++ b/src/recoll_modern/storage/membership.py @@ -4,6 +4,7 @@ import secrets import sqlite3 from dataclasses import dataclass from datetime import UTC, datetime +from pathlib import Path from recoll_modern.storage.database import Database @@ -173,6 +174,19 @@ class MembershipRepository: with self._database.connect() as connection: return _effective(connection, content_hash) + def membership_map_rows(self, document_root: Path) -> list[tuple[str, list[str]]]: + """Return absolute canonical paths with their effective collection tags.""" + with self._database.connect() as connection: + paths = connection.execute( + "SELECT relative_path, document_hash FROM document_paths " + "WHERE is_canonical = 1 AND missing_at IS NULL ORDER BY relative_path" + ).fetchall() + return [ + (str(document_root / str(row[0])), collections) + for row in paths + if (collections := _effective(connection, str(row[1]))) + ] + def _effective(connection: sqlite3.Connection, content_hash: str) -> list[str]: return [ diff --git a/tests/helper/test_membership_helper.py b/tests/helper/test_membership_helper.py new file mode 100644 index 0000000..083b9da --- /dev/null +++ b/tests/helper/test_membership_helper.py @@ -0,0 +1,69 @@ +import os +import subprocess +from pathlib import Path + +import pytest + +from recoll_modern.indexing.map_export import export_membership_map + + +class FakeRows: + def membership_map_rows(self) -> list[tuple[str, list[str]]]: + return [ + ("/documents/a.pdf", ["c0000000002", "c0000000001"]), + ("/documents/odd\nname.pdf", ["c0000000003"]), + ] + + +@pytest.fixture(scope="module") +def compiled_helper(tmp_path_factory: pytest.TempPathFactory) -> Path: + root = Path(__file__).resolve().parents[2] + output = tmp_path_factory.mktemp("helper") / "recoll-membership" + subprocess.run([ + "cc", "-std=c17", "-O2", "-Wall", "-Wextra", "-Werror", + str(root / "helper" / "recoll_membership.c"), "-o", str(output), + ], check=True) + return output + + +def test_helper_returns_sorted_space_separated_ids( + compiled_helper: Path, tmp_path: Path +) -> None: + map_file = tmp_path / "membership.map" + export_membership_map(FakeRows(), map_file) + result = subprocess.run( + [compiled_helper, "/documents/a.pdf"], + env={**os.environ, "RECOLL_MODERN_MEMBERSHIP_MAP": str(map_file)}, + text=True, capture_output=True, check=True, + ) + assert result.stdout == "c0000000001 c0000000002\n" + + +def test_helper_matches_newline_path_and_ignores_unknown( + compiled_helper: Path, tmp_path: Path +) -> None: + map_file = tmp_path / "membership.map" + export_membership_map(FakeRows(), map_file) + env = {**os.environ, "RECOLL_MODERN_MEMBERSHIP_MAP": str(map_file)} + found = subprocess.run( + [compiled_helper, "/documents/odd\nname.pdf"], env=env, + text=True, capture_output=True, check=True, + ) + missing = subprocess.run( + [compiled_helper, "/documents/missing.pdf"], env=env, + text=True, capture_output=True, check=True, + ) + assert found.stdout == "c0000000003\n" + assert missing.stdout == "" + + +def test_helper_rejects_corrupt_map(compiled_helper: Path, tmp_path: Path) -> None: + map_file = tmp_path / "membership.map" + map_file.write_bytes(b"RCM1\xff\xff\xff\xff") + result = subprocess.run( + [compiled_helper, "/documents/a.pdf"], + env={**os.environ, "RECOLL_MODERN_MEMBERSHIP_MAP": str(map_file)}, + text=True, capture_output=True, + ) + assert result.returncode != 0 + assert "corrupt" in result.stderr.lower() diff --git a/tests/indexing/test_job_handlers.py b/tests/indexing/test_job_handlers.py index 6503b46..2c53c88 100644 --- a/tests/indexing/test_job_handlers.py +++ b/tests/indexing/test_job_handlers.py @@ -22,8 +22,14 @@ def _setup( collection_id = SQLiteCollectionRepository(database).create(CollectionCreate("Books")).id jobs = JobRepository(database) membership = MembershipRepository(database) + + class NoopIndexer: + def force_reindex(self, paths: tuple[str, ...]) -> None: + pass + handler = MembershipJobHandler( - PathPolicy(root), DocumentRepository(database), membership, jobs + PathPolicy(root), DocumentRepository(database), membership, jobs, + tmp_path / "membership.map", NoopIndexer(), ) return root, jobs, membership, handler, collection_id @@ -56,3 +62,30 @@ def test_best_effort_applies_valid_files_and_records_errors(tmp_path: Path) -> N assert [(error.relative_path, error.code) for error in jobs.file_errors(job.id)] == [ ("gone.txt", "file_vanished") ] + + +def test_handler_exports_map_before_forced_reindex(tmp_path: Path) -> None: + root, jobs, membership, _, collection_id = _setup(tmp_path) + map_path = tmp_path / "membership.map" + calls: list[tuple[str, ...]] = [] + + class FakeIndexer: + def force_reindex(self, paths: tuple[str, ...]) -> None: + assert map_path.read_bytes().startswith(b"RCM1") + assert collection_id.encode() in map_path.read_bytes() + calls.append(paths) + + database = Database(tmp_path / "state.sqlite3") + handler = MembershipJobHandler( + PathPolicy(root), DocumentRepository(database), membership, jobs, + map_path, FakeIndexer(), + ) + job = jobs.enqueue("membership_add", { + "collection_id": collection_id, + "sources": [{"type": "file", "path": "good.txt"}], + }, False) + claimed = jobs.claim_next() + assert claimed is not None + handler.handle(claimed) + assert calls == [(str(root / "good.txt"),)] + assert jobs.get(job.id).state == JobState.COMPLETED # type: ignore[union-attr] diff --git a/tests/indexing/test_map_export.py b/tests/indexing/test_map_export.py new file mode 100644 index 0000000..2f7adbf --- /dev/null +++ b/tests/indexing/test_map_export.py @@ -0,0 +1,66 @@ +import os +import struct +from pathlib import Path + +import pytest + +from recoll_modern.indexing.map_export import export_membership_map + + +class FakeRows: + def __init__(self, rows: dict[str, list[str]]) -> None: + self.rows = rows + + def membership_map_rows(self) -> list[tuple[str, list[str]]]: + return list(self.rows.items()) + + +def _decode(path: Path) -> dict[str, str]: + data = path.read_bytes() + assert data[:4] == b"RCM1" + offset = 4 + result = {} + while offset < len(data): + path_length = struct.unpack_from(">I", data, offset)[0] + offset += 4 + name = data[offset:offset + path_length].decode() + offset += path_length + value_length = struct.unpack_from(">I", data, offset)[0] + offset += 4 + value = data[offset:offset + value_length].decode() + offset += value_length + result[name] = value + return result + + +def test_export_is_atomic_and_handles_newlines(tmp_path: Path) -> None: + destination = tmp_path / "membership.map" + export_membership_map(FakeRows({"/documents/odd\nname.pdf": ["c0000000001"]}), destination) + assert _decode(destination) == {"/documents/odd\nname.pdf": "c0000000001"} + assert list(tmp_path.glob("*.tmp")) == [] + assert destination.stat().st_mode & 0o777 == 0o640 + + +def test_export_sorts_paths_and_ids(tmp_path: Path) -> None: + destination = tmp_path / "membership.map" + export_membership_map(FakeRows({ + "/documents/z": ["c0000000002", "c0000000001"], + "/documents/a": ["c0000000003"], + }), destination) + assert list(_decode(destination)) == ["/documents/a", "/documents/z"] + assert _decode(destination)["/documents/z"] == "c0000000001 c0000000002" + + +def test_failed_replace_preserves_old_map(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + destination = tmp_path / "membership.map" + export_membership_map(FakeRows({"/documents/old": ["c0000000001"]}), destination) + old = destination.read_bytes() + + def fail_replace(source: str | Path, target: str | Path) -> None: + raise OSError("injected failure") + + monkeypatch.setattr(os, "replace", fail_replace) + with pytest.raises(OSError, match="injected"): + export_membership_map(FakeRows({"/documents/new": ["c0000000002"]}), destination) + assert destination.read_bytes() == old + assert list(tmp_path.glob("*.tmp")) == []