mdcaldav/tests/test_concurrency.py

"""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