diff --git a/src/mdcaldav/index.py b/src/mdcaldav/index.py index cf925bb..2beb60c 100644 --- a/src/mdcaldav/index.py +++ b/src/mdcaldav/index.py @@ -10,6 +10,7 @@ from __future__ import annotations import hashlib import json import sqlite3 +import threading import uuid from dataclasses import dataclass from datetime import datetime, timezone @@ -74,10 +75,21 @@ class Index: self.tasks: dict[str, Task] = {} # uid → current task (in-memory truth) self._parsed: dict[str, list[Task]] = {} # rel_path → tasks + # Every public method takes this. A rescan is a multi-statement + # read-modify-write over a connection shared by Radicale's request + # threads and the watcher thread; interleaving them corrupts the + # reconcile pass and raises sqlite3 API-misuse errors. + self.lock = threading.RLock() + db_path = Path(cfg.index.db).expanduser() db_path.parent.mkdir(parents=True, exist_ok=True) - self.db = sqlite3.connect(db_path, check_same_thread=False) + self.db = sqlite3.connect(db_path, check_same_thread=False, timeout=30.0) self.db.row_factory = sqlite3.Row + # WAL keeps a second process (`mdcaldav doctor` against a running + # server) reading while we write; busy_timeout makes it wait rather + # than fail with "database is locked". + self.db.execute("PRAGMA journal_mode=WAL") + self.db.execute("PRAGMA busy_timeout=30000") self.db.executescript(SCHEMA) if self.db.execute("SELECT COUNT(*) FROM meta").fetchone()[0] == 0: self.db.execute( @@ -102,7 +114,8 @@ class Index: @property def generation(self) -> int: - return self.db.execute("SELECT generation FROM meta").fetchone()[0] + with self.lock: + return self.db.execute("SELECT generation FROM meta").fetchone()[0] def _bump(self) -> int: gen = self.generation + 1 @@ -131,39 +144,40 @@ class Index: def rescan(self, only: list[str] | None = None) -> int: """Reparse the vault (or `only` these files) and reconcile identities.""" - gen = self._bump() - targets = ( - [rel for rel in only if self.included(rel)] - if only is not None - else self.discover_files() - ) - present = set(self.discover_files()) - - pending_new: list[Task] = [] - for rel in targets: - path = self.vault / rel - if not path.exists(): - self._retire_file(rel, gen) - continue - data = path.read_bytes() - parsed = parse(data, rel, self.cfg) - pending_new.extend(self._reconcile(rel, parsed, gen)) - stat = path.stat() - self.db.execute( - "INSERT OR REPLACE INTO files VALUES (?,?,?,?,?)", - (rel, stat.st_mtime, stat.st_size, file_hash(data), gen), + with self.lock: + gen = self._bump() + targets = ( + [rel for rel in only if self.included(rel)] + if only is not None + else self.discover_files() ) + present = set(self.discover_files()) - if only is None: - for row in self.db.execute("SELECT rel_path FROM files").fetchall(): - if row["rel_path"] not in present: - self._retire_file(row["rel_path"], gen) + pending_new: list[Task] = [] + for rel in targets: + path = self.vault / rel + if not path.exists(): + self._retire_file(rel, gen) + continue + data = path.read_bytes() + parsed = parse(data, rel, self.cfg) + pending_new.extend(self._reconcile(rel, parsed, gen)) + stat = path.stat() + self.db.execute( + "INSERT OR REPLACE INTO files VALUES (?,?,?,?,?)", + (rel, stat.st_mtime, stat.st_size, file_hash(data), gen), + ) - self._resolve_moves(pending_new, gen) - self._prune_tombstones(gen) - self.db.commit() - self._rebuild_memory() - return gen + if only is None: + for row in self.db.execute("SELECT rel_path FROM files").fetchall(): + if row["rel_path"] not in present: + self._retire_file(row["rel_path"], gen) + + self._resolve_moves(pending_new, gen) + self._prune_tombstones(gen) + self.db.commit() + self._rebuild_memory() + return gen def refresh_if_stale(self) -> bool: """Rescan only files whose mtime/size no longer match the index. @@ -171,43 +185,56 @@ class Index: Called on every read path so an edit made in vim is visible even when no watcher is running; the watcher (SPEC 8) only improves latency. """ - known = { - row["rel_path"]: (row["mtime"], row["size"]) - for row in self.db.execute("SELECT rel_path, mtime, size FROM files") - } - stale: list[str] = [] - for rel in self.discover_files(): - path = self.vault / rel - try: - stat = path.stat() - except OSError: - continue - if known.pop(rel, None) != (stat.st_mtime, stat.st_size): - stale.append(rel) - stale.extend(known) # files that disappeared - if not stale: - return False - self.rescan(stale) - return True + with self.lock: + known = { + row["rel_path"]: (row["mtime"], row["size"]) + for row in self.db.execute("SELECT rel_path, mtime, size FROM files") + } + stale: list[str] = [] + for rel in self.discover_files(): + path = self.vault / rel + try: + stat = path.stat() + except OSError: + continue + if known.pop(rel, None) != (stat.st_mtime, stat.st_size): + stale.append(rel) + stale.extend(known) # files that disappeared + if not stale: + return False + self.rescan(stale) + return True + + def file_hash_of(self, rel: str) -> str | None: + """The hash the index last recorded for a file, or None if unknown.""" + with self.lock: + row = self.db.execute( + "SELECT hash FROM files WHERE rel_path = ?", (rel,) + ).fetchone() + return row["hash"] if row else None def reassign_uid(self, old_uid: str, new_uid: str) -> None: """Adopt a client-chosen UID for a task we just created.""" if old_uid == new_uid: return - self.db.execute("DELETE FROM tasks WHERE uid = ?", (new_uid,)) - self.db.execute("UPDATE tasks SET uid = ? WHERE uid = ?", (new_uid, old_uid)) - self.db.execute( - "UPDATE tasks SET parent_uid = ? WHERE parent_uid = ?", (new_uid, old_uid) - ) - self.db.commit() - for tasks in self._parsed.values(): - for task in tasks: - if task.uid == old_uid: - task.uid = new_uid - if task.parent_uid == old_uid: - task.parent_uid = new_uid - task.children = [new_uid if c == old_uid else c for c in task.children] - self._rebuild_memory() + with self.lock: + self.db.execute("DELETE FROM tasks WHERE uid = ?", (new_uid,)) + self.db.execute("UPDATE tasks SET uid = ? WHERE uid = ?", (new_uid, old_uid)) + self.db.execute( + "UPDATE tasks SET parent_uid = ? WHERE parent_uid = ?", + (new_uid, old_uid), + ) + self.db.commit() + for tasks in self._parsed.values(): + for task in tasks: + if task.uid == old_uid: + task.uid = new_uid + if task.parent_uid == old_uid: + task.parent_uid = new_uid + task.children = [ + new_uid if c == old_uid else c for c in task.children + ] + self._rebuild_memory() def _retire_file(self, rel: str, gen: int) -> None: self.db.execute( @@ -402,24 +429,29 @@ class Index: # ---- serving --------------------------------------------------------- def _rebuild_memory(self) -> None: - self.tasks = {} - for tasks in self._parsed.values(): - for task in tasks: - self.tasks[task.uid] = task + # Build then swap: assigning an empty dict first would let a reader + # holding no lock observe a half-populated task list. + tasks: dict[str, Task] = {} + for parsed in self._parsed.values(): + for task in parsed: + tasks[task.uid] = task + self.tasks = tasks def completed_at(self, uid: str) -> datetime | None: - row = self.db.execute( - "SELECT completed_at FROM tasks WHERE uid = ?", (uid,) - ).fetchone() + with self.lock: + row = self.db.execute( + "SELECT completed_at FROM tasks WHERE uid = ?", (uid,) + ).fetchone() if row and row["completed_at"]: return datetime.fromisoformat(row["completed_at"]) return None def last_modified(self, uid: str) -> datetime: """When this task last actually changed — the basis of its ETag.""" - row = self.db.execute( - "SELECT last_modified FROM tasks WHERE uid = ?", (uid,) - ).fetchone() + with self.lock: + row = self.db.execute( + "SELECT last_modified FROM tasks WHERE uid = ?", (uid,) + ).fetchone() if row and row["last_modified"]: return datetime.fromisoformat(row["last_modified"]) return datetime.now(timezone.utc) @@ -431,46 +463,51 @@ class Index: if collection is not None: sql += " AND collection = ?" params = (collection,) - row = self.db.execute(sql, params).fetchone() + with self.lock: + row = self.db.execute(sql, params).fetchone() if row and row["m"]: return datetime.fromisoformat(row["m"]) return datetime.now(timezone.utc) def collections(self) -> list[str]: - rows = self.db.execute( - "SELECT DISTINCT collection FROM tasks WHERE deleted_gen IS NULL" - ).fetchall() + with self.lock: + rows = self.db.execute( + "SELECT DISTINCT collection FROM tasks WHERE deleted_gen IS NULL" + ).fetchall() return sorted(r["collection"] for r in rows if r["collection"]) def tasks_in(self, collection: str) -> list[Task]: return [t for t in self.tasks.values() if t.collection == collection] def rel_path_of(self, uid: str) -> str | None: - row = self.db.execute( - "SELECT rel_path FROM tasks WHERE uid = ?", (uid,) - ).fetchone() + with self.lock: + row = self.db.execute( + "SELECT rel_path FROM tasks WHERE uid = ?", (uid,) + ).fetchone() return row["rel_path"] if row else None def sync(self, collection: str, old_token: str = "") -> tuple[str, list[str]]: """Radicale sync-token contract (SPEC 8.1).""" - gen = self.generation - token = f"http://mdcaldav.local/ns/sync/{gen}" - if not old_token: - return token, [f"{t.uid}.ics" for t in self.tasks_in(collection)] + with self.lock: + gen = self.generation + token = f"http://mdcaldav.local/ns/sync/{gen}" + if not old_token: + return token, [f"{t.uid}.ics" for t in self.tasks_in(collection)] - try: - old_gen = int(old_token.rsplit("/", 1)[1]) - except (IndexError, ValueError): - raise ValueError("Malformed sync token") - if old_gen > gen or old_gen < gen - self.cfg.index.tombstone_retention: - raise ValueError("Sync token is too old") + try: + old_gen = int(old_token.rsplit("/", 1)[1]) + except (IndexError, ValueError): + raise ValueError("Malformed sync token") + if old_gen > gen or old_gen < gen - self.cfg.index.tombstone_retention: + raise ValueError("Sync token is too old") - rows = self.db.execute( - "SELECT uid FROM tasks WHERE collection = ?" - " AND (last_modified_gen > ? OR deleted_gen > ?)", - (collection, old_gen, old_gen), - ).fetchall() + rows = self.db.execute( + "SELECT uid FROM tasks WHERE collection = ?" + " AND (last_modified_gen > ? OR deleted_gen > ?)", + (collection, old_gen, old_gen), + ).fetchall() return token, [f"{r['uid']}.ics" for r in rows] def close(self) -> None: - self.db.close() + with self.lock: + self.db.close() diff --git a/src/mdcaldav/writer.py b/src/mdcaldav/writer.py index 8f83ec4..e18c784 100644 --- a/src/mdcaldav/writer.py +++ b/src/mdcaldav/writer.py @@ -66,11 +66,9 @@ class Writer: rel = task.source.rel_path path = self.vault / rel - row = self.index.db.execute( - "SELECT hash FROM files WHERE rel_path = ?", (rel,) - ).fetchone() + known = self.index.file_hash_of(rel) data = path.read_bytes() - if row is None or row["hash"] != file_hash(data): + if known is None or known != file_hash(data): # Changed behind our back — reindex before trusting any span. self.index.rescan([rel]) task = self.index.tasks.get(uid) @@ -96,25 +94,28 @@ class Writer: def apply(self, uid: str, changes: dict) -> None: if not changes: return - task, path, data = self._load(uid) - edits: list[tuple[int, int, bytes]] = [] + # Held across validate → write → reindex: a watcher rescan slipping in + # between would invalidate the byte spans we are about to splice. + with self.index.lock: + task, path, data = self._load(uid) + edits: list[tuple[int, int, bytes]] = [] - if "status" in changes: - marker = self.cfg.tasks.status_marker(changes["status"], task.raw_marker) - edits.append((*task.source.marker_span, marker.encode())) + if "status" in changes: + marker = self.cfg.tasks.status_marker(changes["status"], task.raw_marker) + edits.append((*task.source.marker_span, marker.encode())) - if "title" in changes: - edits.append((*task.source.title_span, changes["title"].encode())) + if "title" in changes: + edits.append((*task.source.title_span, changes["title"].encode())) - if "priority" in changes: - edits.append(self._priority_edit(task, changes["priority"])) + if "priority" in changes: + edits.append(self._priority_edit(task, changes["priority"])) - if "description" in changes: - edits.append(self._description_edit(task, changes["description"], data)) + if "description" in changes: + edits.append(self._description_edit(task, changes["description"], data)) - self._backup(task.source.rel_path, path) - atomic_write(path, splice(data, [e for e in edits if e is not None])) - self.index.rescan([task.source.rel_path]) + self._backup(task.source.rel_path, path) + atomic_write(path, splice(data, [e for e in edits if e is not None])) + self.index.rescan([task.source.rel_path]) def _priority_edit(self, task: Task, priority: int | None) -> tuple[int, int, bytes]: letter = self.cfg.tasks.priority_letter(priority) @@ -168,62 +169,64 @@ class Writer: if not self.cfg.write.allow_create: raise NotAllowedError("task creation is disabled") - rel = self.inbox_for(collection) - path = self.vault / rel - heading = self.cfg.write.inbox_heading - marker = self.cfg.tasks.status_marker( - fields.get("status", Status.NEEDS_ACTION), None - ) - letter = self.cfg.tasks.priority_letter(fields.get("priority")) - prefix = f"[#{letter}] " if letter else "" - line = f"- [{marker}] {prefix}{fields.get('title', 'Untitled')}" + with self.index.lock: + rel = self.inbox_for(collection) + path = self.vault / rel + heading = self.cfg.write.inbox_heading + marker = self.cfg.tasks.status_marker( + fields.get("status", Status.NEEDS_ACTION), None + ) + letter = self.cfg.tasks.priority_letter(fields.get("priority")) + prefix = f"[#{letter}] " if letter else "" + line = f"- [{marker}] {prefix}{fields.get('title', 'Untitled')}" - if path.exists(): - text = path.read_text(encoding="utf-8") - if heading in text: - head, _, tail = text.partition(heading) - body = f"{head}{heading}\n{line}\n{tail.lstrip(chr(10))}" + if path.exists(): + text = path.read_text(encoding="utf-8") + if heading in text: + head, _, tail = text.partition(heading) + body = f"{head}{heading}\n{line}\n{tail.lstrip(chr(10))}" + else: + sep = "" if text.endswith("\n") else "\n" + body = f"{text}{sep}\n{heading}\n\n{line}\n" else: - sep = "" if text.endswith("\n") else "\n" - body = f"{text}{sep}\n{heading}\n\n{line}\n" - else: - path.parent.mkdir(parents=True, exist_ok=True) - body = f"{heading}\n\n{line}\n" + path.parent.mkdir(parents=True, exist_ok=True) + body = f"{heading}\n\n{line}\n" - if path.exists(): - self._backup(rel, path) - atomic_write(path, body.encode()) - else: - path.write_bytes(body.encode()) + if path.exists(): + self._backup(rel, path) + atomic_write(path, body.encode()) + else: + path.write_bytes(body.encode()) - self.index.rescan([rel]) - title = fields.get("title", "Untitled") - for task in self.index.tasks.values(): - if task.source.rel_path == rel and task.title == title: - uid = task.uid - # Adopt the client's UID so the item stays at the href it PUT to. - if wanted := fields.get("uid"): - self.index.reassign_uid(uid, wanted) - uid = wanted - if desc := fields.get("description"): - self.apply(uid, {"description": desc}) - return uid - raise RuntimeError("created task did not survive reindex") + self.index.rescan([rel]) + title = fields.get("title", "Untitled") + for task in self.index.tasks.values(): + if task.source.rel_path == rel and task.title == title: + uid = task.uid + # Adopt the client's UID so the item stays at the href it PUT to. + if wanted := fields.get("uid"): + self.index.reassign_uid(uid, wanted) + uid = wanted + if desc := fields.get("description"): + self.apply(uid, {"description": desc}) + return uid + raise RuntimeError("created task did not survive reindex") def delete(self, uid: str) -> None: if not self.cfg.write.allow_delete: raise NotAllowedError("task deletion is disabled") - task, path, data = self._load(uid) + with self.index.lock: + task, path, data = self._load(uid) - policy = self.cfg.write.delete_children - if task.children and policy == "reject": - raise NotAllowedError("task has subtasks and delete_children=reject") + policy = self.cfg.write.delete_children + if task.children and policy == "reject": + raise NotAllowedError("task has subtasks and delete_children=reject") - start = _back_over_newline(data, task.source.line_span[0]) - end = self._subtree_end(task) if policy == "cascade" else self._own_end(task) - self._backup(task.source.rel_path, path) - atomic_write(path, splice(data, [(start, end, b"")])) - self.index.rescan([task.source.rel_path]) + start = _back_over_newline(data, task.source.line_span[0]) + end = self._subtree_end(task) if policy == "cascade" else self._own_end(task) + self._backup(task.source.rel_path, path) + atomic_write(path, splice(data, [(start, end, b"")])) + self.index.rescan([task.source.rel_path]) def _own_end(self, task: Task) -> int: end = task.source.line_span[1] diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py new file mode 100644 index 0000000..9947075 --- /dev/null +++ b/tests/test_concurrency.py @@ -0,0 +1,147 @@ +"""The watcher thread and Radicale's request threads share one Index. + +Radicale serializes request threads with `Storage.acquire_lock`, but the watcher +rescans outside it. Unsynchronized, the two interleave inside `rescan`'s +multi-statement read-modify-write and raise `sqlite3.OperationalError: database +is locked`, `sqlite3.InterfaceError: bad parameter or other API misuse`, or a +`TypeError` from a row that vanished mid-pass. +""" + +from __future__ import annotations + +import threading +import time +from pathlib import Path + +from mdcaldav.config import Config +from mdcaldav.index import Index +from mdcaldav.writer import Writer +from mdcaldav.model import Status + +DURATION = 3.0 + + +def _index(vault: Path, tmp_path: Path) -> Index: + cfg = Config() + cfg.vault.path = vault + cfg.index.db = tmp_path / "index.db" + index = Index(cfg) + index.rescan() + return index + + +def _run(workers: list[threading.Thread], stop: threading.Event) -> None: + for worker in workers: + worker.start() + time.sleep(DURATION) + stop.set() + for worker in workers: + worker.join(timeout=10) + assert not worker.is_alive(), "worker did not stop" + + +def test_concurrent_rescans_do_not_corrupt_the_index(vault: Path, tmp_path: Path): + index = _index(vault, tmp_path) + note = vault / "daily" / "2026-01-05.md" + errors: list[BaseException] = [] + stop = threading.Event() + + def churn() -> None: + n = 0 + while not stop.is_set(): + note.write_bytes(note.read_bytes() + f"\n- [ ] churn {n}\n".encode()) + n += 1 + time.sleep(0.005) + + def rescanner() -> None: + while not stop.is_set(): + try: + index.rescan(["daily/2026-01-05.md"]) + index.refresh_if_stale() + index.collections() + except BaseException as exc: # noqa: BLE001 - the assertion is "none" + errors.append(exc) + return + + workers = [threading.Thread(target=churn, daemon=True)] + workers += [threading.Thread(target=rescanner, daemon=True) for _ in range(4)] + _run(workers, stop) + + assert not errors, f"{len(errors)} failures, first: {errors[0]!r}" + + +def test_writes_survive_a_concurrent_watcher(vault: Path, tmp_path: Path): + """Completing tasks while the watcher rescans must not lose or garble edits.""" + index = _index(vault, tmp_path) + writer = Writer(index.cfg, index) + other = vault / "daily" / "2026-01-06.md" + errors: list[BaseException] = [] + completed: list[str] = [] + stop = threading.Event() + + def watcher() -> None: + # Stands in for the watchdog thread: touch a file, then reindex it. + while not stop.is_set(): + try: + other.write_bytes(other.read_bytes()) + index.rescan(["daily/2026-01-06.md"]) + except BaseException as exc: # noqa: BLE001 + errors.append(exc) + return + time.sleep(0.01) + + def completer() -> None: + while not stop.is_set(): + pending = [ + t + for t in index.tasks.values() + if t.status is Status.NEEDS_ACTION + and t.source.rel_path == "daily/2026-01-07.md" + ] + if not pending: + return + task = pending[0] + try: + writer.apply(task.uid, {"status": Status.COMPLETED}) + completed.append(task.title) + except BaseException as exc: # noqa: BLE001 + errors.append(exc) + return + time.sleep(0.01) + + _run([threading.Thread(target=f, daemon=True) for f in (watcher, completer)], stop) + + assert not errors, f"{len(errors)} failures, first: {errors[0]!r}" + assert completed, "no task was completed" + + text = (vault / "daily/2026-01-07.md").read_text() + for title in completed: + assert f"] {title}" in text # the line still parses as a task + assert text.count("- [x]") >= len(completed) + + +def test_etag_input_is_stable_under_a_concurrent_watcher(vault: Path, tmp_path: Path): + """An idle rescan loop must not change any task's last_modified. + + This is what the ETag hashes; churn here is what made clients revert their + own edits. + """ + index = _index(vault, tmp_path) + uids = list(index.tasks) + before = {uid: index.last_modified(uid) for uid in uids} + errors: list[BaseException] = [] + stop = threading.Event() + + def rescanner() -> None: + while not stop.is_set(): + try: + index.rescan() + except BaseException as exc: # noqa: BLE001 + errors.append(exc) + return + time.sleep(0.01) + + _run([threading.Thread(target=rescanner, daemon=True) for _ in range(3)], stop) + + assert not errors, f"{len(errors)} failures, first: {errors[0]!r}" + assert {uid: index.last_modified(uid) for uid in uids} == before