feat: export Recoll membership metadata

This commit is contained in:
Lucas Winther
2026-09-26 10:51:40 +02:00
parent cbc07b2897
commit f6f03b9128
12 changed files with 391 additions and 16 deletions
+1
View File
@@ -30,6 +30,7 @@ htmlcov/
*.sqlite3-shm
*.sqlite3-wal
membership.map
helper/recoll-membership
# Recoll runtime data (configuration templates remain tracked)
recoll.conf
+8 -4
View File
@@ -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
+4 -5
View File
@@ -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:
+5 -5
View File
@@ -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`
+12
View File
@@ -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
+90
View File
@@ -0,0 +1,90 @@
#define _POSIX_C_SOURCE 200809L
#include <errno.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#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;
}
+20 -1
View File
@@ -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))
+68
View File
@@ -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)
+14
View File
@@ -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 [
+69
View File
@@ -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()
+34 -1
View File
@@ -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]
+66
View File
@@ -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")) == []