diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 605a7530..b00fa2b8 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -623,7 +623,12 @@ The tool implements a SQLite-based caching system for performance: turns through a fair cross-process queue (`write_queue.py`), so none gives up on a sibling. Workers write only rows: an existing search index is caught up once at the end of the pass, by the parent, not inside each - worker's write transaction. Against other processes connections wait 30 s + worker's write transaction. One pass at a time per cache file: a pass + holds a lock (`.pass-lock`, `pass_lock.py`) for its whole run, + catch-up included, so a second `claude-code-log` or a `serve --watch` + tick over the same archive waits for it instead of contending + (`CLAUDE_CODE_LOG_PASS_LOCK=` bounds the wait, `=0` disables + the lock). Against other processes connections wait 30 s (`CLAUDE_CODE_LOG_CACHE_TIMEOUT` overrides, in seconds), and a project locked out is retried after the pool rather than failed. A pass compacts an oversized WAL before it starts and truncates the WAL when it ends. A diff --git a/claude_code_log/converter.py b/claude_code_log/converter.py index 67ff583c..7a306b06 100644 --- a/claude_code_log/converter.py +++ b/claude_code_log/converter.py @@ -6584,14 +6584,26 @@ def process_projects_hierarchy( it, so a caller can tell a complete pass from one that lost projects (the CLI exits non-zero on the latter). """ + from .pass_lock import pass_lock from .search import deferred_index_maintenance + # One pass at a time per cache file (see `pass_lock`): a second pass, + # or a `serve --watch` tick beside a CLI run, waits here instead of + # competing for the write lock with a queue of its own. Taken before + # the lease and the WAL trim, released after the index catch-up and + # the checkpoint below, so the waiting pass never contends with the + # serial catch-up and starts on a current index and a reset log. + lock = ( + pass_lock(get_cache_db_path(projects_path)) + if use_cache + else contextlib.nullcontext() + ) lease = ( connection_lease(get_cache_db_path(projects_path)) if use_cache else contextlib.nullcontext() ) - with lease: + with lock, lease: if use_cache: # Inside the lease, so the connection that recovers the log # is the one the pass then keeps. diff --git a/claude_code_log/pass_lock.py b/claude_code_log/pass_lock.py new file mode 100644 index 00000000..76835cf4 --- /dev/null +++ b/claude_code_log/pass_lock.py @@ -0,0 +1,215 @@ +"""One hierarchy pass per cache file. + +Inside a pass, the project workers share a fair write queue +(`write_queue`), so none of them gives up on a sibling. Two *passes* over +the same cache — a second ``claude-code-log`` run started while one is +going, or ``serve --watch`` ticking while a CLI run refreshes the same +archive — share nothing: each is, to the other, one tight-loop writer, +and SQLite's lock polling is unfair to outsiders (a writer that just +committed tends to re-acquire before a sleeper wakes), so the second pass +can lose projects to ``database is locked`` however patient its own +workers are. It is also the realistic way two refreshes of one project +overlap and clear each other's ``refresh_pending``. + +So a pass takes the **pass lock** first: an exclusive transaction on a +sidecar SQLite file next to the cache (``.pass-lock``), held until +the pass — its WAL trim, its conversions, its checkpoint — is over. +SQLite supplies the cross-platform advisory lock and releases it with +the holder's process, so a killed pass leaves nothing stale to clean up. +Nothing else opens that file. A second pass waits, says so after a +moment, and then runs — usually to find the work done. The wait is +unbounded by default; ``CLAUDE_CODE_LOG_PASS_LOCK=`` bounds it +(a pass that runs out of patience fails with `PassLockTimeout`), and +``=0`` disables the lock altogether. + +Re-entrant within a process — process-wide, not per thread: a pass +started while this process already holds the lock does not wait on +itself. Nothing nests today (``serve`` finishes its startup pass before +its watch thread starts ticking), so this is a guard against a future +caller, not something the server relies on. The TUI's writes and a +single-file ``convert_jsonl_to`` are short transactions outside any +pass; they keep the ordinary busy timeout and are not gated here. + +The lock degrades rather than fails: when the lock file cannot be +created (a read-only directory), is not a SQLite file (junk written +there), or cannot be written (another user's file — a read-only +``BEGIN EXCLUSIVE`` excludes nobody, so holding one would only pretend), +the pass says so, names the file, and runs without the lock. +""" + +from __future__ import annotations + +import os +import sqlite3 +import threading +import time +from contextlib import contextmanager +from pathlib import Path +from typing import Callable, Dict, Iterator, Optional, Tuple + +#: ``0``/``off`` disables the pass lock; a number bounds the wait in seconds. +ENV_PASS_LOCK = "CLAUDE_CODE_LOG_PASS_LOCK" + +#: Seconds of waiting before the pass says what it is waiting for. +DEFAULT_NOTICE_AFTER = 2.0 + +_POLL_SECONDS = 0.25 + + +class PassLockTimeout(RuntimeError): + """Another pass held the cache for longer than this one may wait.""" + + +def pass_lock_path(db_path: Path) -> Path: + """The lock file for ``db_path``: beside it, ``.pass-lock``. + + Keyed on the resolved path, so a cache reached through a symlink has + the same lock as the cache reached directly. + """ + db_path = Path(db_path).resolve() + return db_path.with_name(db_path.name + ".pass-lock") + + +def _wait_budget() -> Tuple[bool, Optional[float]]: + """``(enabled, seconds)``: seconds is None for an unbounded wait.""" + raw = os.getenv(ENV_PASS_LOCK, "") + value = raw.strip().lower() + if not value: + # Unset, or set to nothing (`VAR= command`): the default, like every + # other CLAUDE_CODE_LOG_* switch. + return True, None + if value in ("off", "false", "no"): + return False, None + try: + seconds = float(value) + except ValueError: + raise ValueError( + f"{ENV_PASS_LOCK}={raw!r}: expected a number of seconds, or 0 to " + "disable the pass lock" + ) from None + if seconds <= 0: + return False, None + return True, seconds + + +# This process's held locks: lock file -> (connection, re-entry depth). +_holders: Dict[str, Tuple[sqlite3.Connection, int]] = {} +_guard = threading.Lock() + + +def pass_lock_held(db_path: Path) -> bool: + """Whether this process holds the pass lock for ``db_path``'s cache. + + For tests pinning what runs under the lock; a disabled or degraded + lock is never "held". + """ + with _guard: + return str(pass_lock_path(db_path)) in _holders + + +def _acquire( + path: Path, + budget: Optional[float], + notice_after: float, + announce: Callable[[str], None], +) -> sqlite3.Connection: + from .cache import is_busy_database_error + + # `timeout=0`: SQLite's own busy handler would wait too, but silently; + # polling here is what lets the pass say what it is waiting for. + conn = sqlite3.connect(path, timeout=0, isolation_level=None) + started = time.monotonic() + announced = False + try: + while True: + try: + conn.execute("BEGIN EXCLUSIVE") + # A file this process cannot write opens read-only, and a + # read-only exclusive transaction takes no lock at all. One + # write inside the transaction makes that surface here, as + # `attempt to write a readonly database`; it is rolled back + # with the lock, so the file stays empty. + conn.execute("PRAGMA user_version = 1") + return conn + except sqlite3.OperationalError as exc: + if not is_busy_database_error(exc): + raise + waited = time.monotonic() - started + if budget is not None and waited >= budget: + raise PassLockTimeout( + f"another claude-code-log pass has held {path} for the " + f"{budget:g} s this one was allowed to wait " + f"({ENV_PASS_LOCK}); nothing was converted" + ) + if not announced and waited >= notice_after: + announce( + "another claude-code-log pass is running on this cache " + f"({path}); waiting for it to finish" + ) + announced = True + remaining = _POLL_SECONDS if budget is None else budget - waited + time.sleep(max(0.0, min(_POLL_SECONDS, remaining))) + except BaseException: + conn.close() + raise + + +@contextmanager +def pass_lock( + db_path: Path, + *, + notice_after: float = DEFAULT_NOTICE_AFTER, + announce: Callable[[str], None] = print, +) -> Iterator[None]: + """Hold the pass lock for ``db_path``'s cache for the scope. + + Waits for another process's pass to finish (announcing that after + ``notice_after`` seconds), re-enters freely within this process, and + is a no-op when disabled by the environment — or when the lock file + cannot be used (not creatable, not a database, not writable), which + is announced rather than allowed to fail a pass that could otherwise + run. + """ + enabled, budget = _wait_budget() + if not enabled: + yield + return + path = pass_lock_path(db_path) + key = str(path) + + with _guard: + held = _holders.get(key) + if held is not None: + _holders[key] = (held[0], held[1] + 1) + if held is not None: + try: + yield + finally: + with _guard: + conn, depth = _holders[key] + _holders[key] = (conn, depth - 1) + return + + try: + conn = _acquire(path, budget, notice_after, announce) + except sqlite3.DatabaseError as exc: + # OperationalError (cannot create or write the file) and its parent + # DatabaseError (`file is not a database`): without the lock, not + # without the pass. + announce( + f"Warning: could not take the pass lock at {path} ({exc}); " + "running without it" + ) + yield + return + with _guard: + _holders[key] = (conn, 1) + try: + yield + finally: + with _guard: + _holders.pop(key, None) + try: + conn.rollback() + finally: + conn.close() diff --git a/dev-docs/application_model.md b/dev-docs/application_model.md index 0e4b9c58..cc72896e 100644 --- a/dev-docs/application_model.md +++ b/dev-docs/application_model.md @@ -215,6 +215,41 @@ in the parent) is the open lever if that worst case ever matters. A single file or directory conversion (`convert_jsonl_to` with no pass around it) still re-indexes inside the write, where nothing competes for the lock. `CLAUDE_CODE_LOG_SEARCH_AUTO_INDEX=0` switches both off. +The queue is per pass, so two *passes* over one cache — a second CLI +run, or a `serve --watch` tick while a CLI run refreshes the same +archive — are, to each other, one tight-loop writer each, and the +unfair polling above applies between them (it is also the realistic +way two refreshes of one project overlap and clear each other's +`refresh_pending`). So `process_projects_hierarchy` first takes the +**pass lock** ([`pass_lock.py`](../claude_code_log/pass_lock.py)): an +exclusive transaction on a sidecar SQLite file beside the cache +(`.pass-lock`), held around the WAL trim, the conversions, the +index catch-up and the checkpoint, and released with the holder's +process (nothing stale to clean after a kill). A second pass waits, +announces what it is waiting for after 2 s, then runs and usually +finds the work done; +`CLAUDE_CODE_LOG_PASS_LOCK=` bounds the wait (then +`PassLockTimeout`, nothing converted) and `=0` disables the lock. +Re-entrant within a process (process-wide, not per thread), though +nothing nests today: `serve` finishes its startup pass before its watch +thread ticks. The lock is keyed on the resolved cache path, so a +symlinked cache shares it. It degrades rather than fails: a lock file +that cannot be created (read-only directory), is not SQLite (junk), or +cannot be written (another user's; a read-only `BEGIN EXCLUSIVE` +excludes nobody, which one rolled-back write inside the transaction +makes surface) is announced by name and the pass runs without the +lock. Outside the gate, by design: the TUI's writes and a +single-file `convert_jsonl_to` are short transactions with the +ordinary busy timeout, as before. `test_pass_lock.py` pins the +waiting, the bound, the switch, the three degrade cases, the symlink, +two concurrent passes serialising, and that the lock is held through +the WAL trim, the index catch-up and the checkpoint +(`pass_lock.pass_lock_held`, recorded from inside each). +Measured on the 32 largest projects of a real archive (every file +re-saved, two passes started together at `--jobs 8`): without the lock +each pass took ~872 s and the pair re-saved 1.4x the files; with it the +first ran alone in 552 s and the second, after waiting, found every +project current in 0.2 s. `serve` takes the same `--jobs` as the main command. A pool-level failure (e.g. a library caller without the `if __name__ == "__main__"` guard `spawn` needs) degrades to inline sequential processing with a warning rather than aborting. diff --git a/docs/live-updates.md b/docs/live-updates.md index 9f110316..fec04a49 100644 --- a/docs/live-updates.md +++ b/docs/live-updates.md @@ -114,6 +114,13 @@ page notices on its next poll. Raise `--quiet-period` if conversions feel too frequent on a large project; lower `--interval` if you want changes noticed sooner. +A `serve --watch` or `watch --all-projects` tick and a plain +`claude-code-log` run over the same archive take turns rather than +racing for the cache: whichever starts second waits for the first to +finish (and says so after a couple of seconds), then runs — usually to +find the work already done. `CLAUDE_CODE_LOG_PASS_LOCK=` bounds +that wait; `CLAUDE_CODE_LOG_PASS_LOCK=0` turns the lock off. + Switching `--theme` rewrites every page in the new theme on the first tick — including a combined page that `--combined no` would otherwise leave alone, so its session pages never link back into a page of the diff --git a/test/test_cache_sqlite_integrity.py b/test/test_cache_sqlite_integrity.py index d5b333e2..d3ac8641 100644 --- a/test/test_cache_sqlite_integrity.py +++ b/test/test_cache_sqlite_integrity.py @@ -699,6 +699,12 @@ class TestConnectionCensus: 1, "mode=ro reader: one SELECT across projects, then close", ), + ("pass_lock.py", "_acquire"): ( + 1, + "the sidecar `.pass-lock` file, not the cache database: " + "holds `BEGIN EXCLUSIVE` as the pass lock; its one write (a " + "`user_version` probe that surfaces a read-only file) is rolled back", + ), } @staticmethod diff --git a/test/test_pass_lock.py b/test/test_pass_lock.py new file mode 100644 index 00000000..1a8bac95 --- /dev/null +++ b/test/test_pass_lock.py @@ -0,0 +1,412 @@ +"""One hierarchy pass per cache file: `pass_lock`. + +Two passes over the same cache — a second CLI run, or `serve --watch` +ticking while a CLI pass runs — write through two separate ticket +queues, so to each other they are one tight-loop writer and SQLite's +unfair lock polling applies (issue #332's mechanism). The pass lock makes +the second pass wait for the first instead. +""" + +from __future__ import annotations + +import multiprocessing +import os +import sqlite3 +import sys +import time +from concurrent.futures import ProcessPoolExecutor +from pathlib import Path +from typing import Any + +import pytest + +from claude_code_log.pass_lock import ( + ENV_PASS_LOCK, + PassLockTimeout, + _wait_budget, + pass_lock, + pass_lock_path, +) + +_released: Any = None +_held: Any = None + + +def _init(released: Any, held: Any) -> None: + global _released, _held + _released = released + _held = held + + +def _hold(db: str, env: dict[str, str]) -> str: + """Hold the pass lock on ``db`` until the parent sets ``_released``.""" + os.environ.update(env) + with pass_lock(Path(db)): + _held.set() + _released.wait(60) + return "released" + + +def _spawn_holder(db_path: Path, env: dict[str, str] | None = None) -> Any: + ctx = multiprocessing.get_context("spawn") + released = ctx.Event() + held = ctx.Event() + pool = ProcessPoolExecutor( + max_workers=1, mp_context=ctx, initializer=_init, initargs=(released, held) + ) + future = pool.submit(_hold, str(db_path), env or {}) + assert held.wait(60), "the holder never took the lock" + return pool, released, future + + +def test_a_second_process_waits_for_the_pass_lock( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + db_path = tmp_path / "cache.db" + pool, released, future = _spawn_holder(db_path) + try: + monkeypatch.setenv(ENV_PASS_LOCK, "0.3") + started = time.monotonic() + with pytest.raises(PassLockTimeout) as excinfo: + with pass_lock(db_path): + pass + assert time.monotonic() - started >= 0.3 + assert str(pass_lock_path(db_path)) in str(excinfo.value) + + released.set() + assert future.result(timeout=60) == "released" + monkeypatch.delenv(ENV_PASS_LOCK) + with pass_lock(db_path): + pass # free again, and the default (unbounded wait) returns at once + finally: + released.set() + pool.shutdown(wait=True) + + +def test_the_pass_lock_is_reentrant_in_a_process(tmp_path: Path) -> None: + db_path = tmp_path / "cache.db" + with pass_lock(db_path): + with pass_lock(db_path): + pass + # Still held by the outer scope: another process cannot take it. + conn = sqlite3.connect(pass_lock_path(db_path), timeout=0.1) + try: + with pytest.raises(sqlite3.OperationalError): + conn.execute("BEGIN EXCLUSIVE") + finally: + conn.close() + # Released with the outer scope. + conn = sqlite3.connect(pass_lock_path(db_path), timeout=0.1) + try: + conn.execute("BEGIN EXCLUSIVE") + conn.rollback() + finally: + conn.close() + + +def test_the_pass_lock_can_be_switched_off( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """``CLAUDE_CODE_LOG_PASS_LOCK=0`` restores the pre-lock behaviour: + nothing is taken, so the file is never created and nobody waits.""" + db_path = tmp_path / "cache.db" + monkeypatch.setenv(ENV_PASS_LOCK, "0") + pool, released, future = _spawn_holder(db_path, {ENV_PASS_LOCK: "0"}) + try: + with pass_lock(db_path): + pass + assert not pass_lock_path(db_path).exists() + finally: + released.set() + future.result(timeout=60) + pool.shutdown(wait=True) + + +@pytest.mark.parametrize( + ("value", "enabled", "budget"), + [ + (None, True, None), + ("", True, None), + (" ", True, None), + ("0", False, None), + ("off", False, None), + ("no", False, None), + ("false", False, None), + ("-1", False, None), + ("2.5", True, 2.5), + (" 30 ", True, 30.0), + ], +) +def test_the_environment_sets_the_wait_budget( + monkeypatch: pytest.MonkeyPatch, + value: str | None, + enabled: bool, + budget: float | None, +) -> None: + """Unset or empty means the default (the lock, an unbounded wait): an + empty value is what ``VAR= command`` and a blank config line produce, + and every other ``CLAUDE_CODE_LOG_*`` switch reads it as unset. ``0`` + and the usual spellings of "off" disable the lock; a positive number + bounds the wait.""" + if value is None: + monkeypatch.delenv(ENV_PASS_LOCK, raising=False) + else: + monkeypatch.setenv(ENV_PASS_LOCK, value) + assert _wait_budget() == (enabled, budget) + + +def test_an_unreadable_wait_budget_is_refused(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(ENV_PASS_LOCK, "soon") + with pytest.raises(ValueError, match="CLAUDE_CODE_LOG_PASS_LOCK='soon'"): + _wait_budget() + + +def test_waiting_is_announced_with_the_lock_file( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + db_path = tmp_path / "cache.db" + pool, released, future = _spawn_holder(db_path) + try: + monkeypatch.setenv(ENV_PASS_LOCK, "1.0") + with pytest.raises(PassLockTimeout): + with pass_lock(db_path, notice_after=0.2): + pass + finally: + released.set() + future.result(timeout=60) + pool.shutdown(wait=True) + out = capsys.readouterr().out + assert "another claude-code-log pass" in out + assert str(pass_lock_path(db_path)) in out + + +def test_a_symlinked_cache_shares_its_lock( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The lock is keyed on the resolved cache path: a pass reaching the + cache through a symlink waits on the pass holding it by its real path, + instead of taking a lock of its own beside the link.""" + real = tmp_path / "real" + real.mkdir() + db_path = real / "cache.db" + db_path.write_bytes(b"") + link_dir = tmp_path / "link" + try: + link_dir.symlink_to(real, target_is_directory=True) + except (OSError, NotImplementedError) as exc: + pytest.skip(f"cannot create a symlink here: {exc}") + linked_db = link_dir / "cache.db" + assert pass_lock_path(linked_db) == pass_lock_path(db_path) + + pool, released, future = _spawn_holder(db_path) + try: + monkeypatch.setenv(ENV_PASS_LOCK, "0.3") + with pytest.raises(PassLockTimeout): + with pass_lock(linked_db): + pass + finally: + released.set() + future.result(timeout=60) + pool.shutdown(wait=True) + + +_not_root = pytest.mark.skipif( + hasattr(os, "geteuid") and os.geteuid() == 0, # pragma: no cover + reason="root ignores file modes", +) + + +def _degraded(announced: list[str], path: Path) -> bool: + return any("running without it" in m and str(path) in m for m in announced) + + +def test_a_lock_file_that_is_not_a_database_is_announced_and_bypassed( + tmp_path: Path, +) -> None: + """Junk in `.pass-lock` must not fail every pass from then on: + the pass says so, names the file, and runs without the lock.""" + db_path = tmp_path / "cache.db" + lock_path = pass_lock_path(db_path) + lock_path.write_bytes(b"not a database\n" * 10) + announced: list[str] = [] + with pass_lock(db_path, announce=announced.append): + pass + assert _degraded(announced, lock_path), announced + + +@_not_root +def test_a_read_only_lock_file_is_announced_and_bypassed(tmp_path: Path) -> None: + """A lock file this process cannot write (another user's, say) opens + read-only, and a read-only `BEGIN EXCLUSIVE` excludes nobody — so the + pass must not pretend to hold the lock: it announces and runs without.""" + db_path = tmp_path / "cache.db" + lock_path = pass_lock_path(db_path) + with pass_lock(db_path): + pass # creates the (empty) lock file + lock_path.chmod(0o444) + try: + announced: list[str] = [] + with pass_lock(db_path, announce=announced.append): + pass + assert _degraded(announced, lock_path), announced + finally: + lock_path.chmod(0o644) + + +@_not_root +@pytest.mark.skipif(sys.platform == "win32", reason="directory modes are POSIX") +def test_a_read_only_directory_is_announced_and_bypassed(tmp_path: Path) -> None: + """No lock file can be created beside the cache: announced, bypassed.""" + cache_dir = tmp_path / "ro" + cache_dir.mkdir() + db_path = cache_dir / "cache.db" + cache_dir.chmod(0o555) + try: + announced: list[str] = [] + with pass_lock(db_path, announce=announced.append): + pass + assert _degraded(announced, pass_lock_path(db_path)), announced + assert not pass_lock_path(db_path).exists() + finally: + cache_dir.chmod(0o755) + + +def _entry_line(text: str) -> str: + import json + + return ( + json.dumps( + { + "type": "user", + "timestamp": "2026-01-01T00:00:00Z", + "parentUuid": None, + "isSidechain": False, + "userType": "human", + "cwd": "/tmp", + "sessionId": "sess-lock", + "version": "1.0.0", + "uuid": "uuid-lock", + "message": { + "role": "user", + "content": [{"type": "text", "text": text}], + }, + } + ) + + "\n" + ) + + +def test_a_hierarchy_pass_takes_the_pass_lock( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """While another process holds the lock, the pass waits — and with a + bounded wait, fails without converting anything.""" + from claude_code_log.cache import get_cache_db_path + from claude_code_log.converter import process_projects_hierarchy + + projects = tmp_path / "projects" + project = projects / "-home-u-proj" + project.mkdir(parents=True) + (project / "session.jsonl").write_text(_entry_line("hello")) + db_path = get_cache_db_path(projects) + + pool, released, future = _spawn_holder(db_path) + try: + monkeypatch.setenv(ENV_PASS_LOCK, "0.3") + with pytest.raises(PassLockTimeout): + process_projects_hierarchy(projects, silent=True) + assert not (projects / "index.html").exists(), "converted without the lock" + + released.set() + assert future.result(timeout=60) == "released" + finally: + released.set() + pool.shutdown(wait=True) + + monkeypatch.delenv(ENV_PASS_LOCK) + process_projects_hierarchy(projects, silent=True) + assert (projects / "index.html").exists() + + +def _run_pass(projects: str) -> tuple[float, float] | None: + """Run one hierarchy pass; return the wall-clock window during which it + held the pass lock (None if it never took one).""" + from contextlib import contextmanager + + import claude_code_log.pass_lock as pass_lock_module + from claude_code_log.converter import process_projects_hierarchy + + real = pass_lock_module.pass_lock + window: list[float] = [] + + @contextmanager + def timed(db_path: Path, **kwargs: Any): + with real(db_path, **kwargs): + window.append(time.time()) + try: + yield + finally: + window.append(time.time()) + + pass_lock_module.pass_lock = timed # the converter imports it per call + process_projects_hierarchy(Path(projects), silent=True) + return (window[0], window[-1]) if window else None + + +def test_two_passes_over_one_cache_run_one_after_the_other(tmp_path: Path) -> None: + """Started together, the two passes hold the cache one after the other.""" + projects = tmp_path / "projects" + for name in ("-home-u-a", "-home-u-b"): + project = projects / name + project.mkdir(parents=True) + (project / "session.jsonl").write_text(_entry_line(f"text of {name}")) + + ctx = multiprocessing.get_context("spawn") + with ProcessPoolExecutor(max_workers=2, mp_context=ctx) as pool: + futures = [pool.submit(_run_pass, str(projects)) for _ in range(2)] + results = [f.result(timeout=120) for f in futures] + assert all(results), f"a pass ran without the lock: {results}" + windows = sorted(w for w in results if w is not None) + (_, end_first), (start_second, _) = windows + assert end_first <= start_second, f"passes overlapped: {windows}" + assert (projects / "index.html").exists() + + +def test_the_pass_lock_covers_the_trim_the_catch_up_and_the_checkpoint( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The lock is taken before the WAL trim and released only after the + search-index catch-up and the WAL checkpoint, so a waiting pass never + contends with the serial catch-up and starts on a reset log.""" + from claude_code_log import converter + from claude_code_log.cache import get_cache_db_path + from claude_code_log.converter import process_projects_hierarchy + from claude_code_log.pass_lock import pass_lock_held + + projects = tmp_path / "projects" + project = projects / "-home-u-proj" + project.mkdir(parents=True) + (project / "session.jsonl").write_text(_entry_line("hello")) + db_path = get_cache_db_path(projects) + + held: dict[str, bool] = {} + for name in ("trim_oversized_wal", "catch_up_search_index", "checkpoint_wal"): + real = getattr(converter, name) + + def recording( + path: Path, *args: Any, _name: str = name, _real: Any = real, **kwargs: Any + ) -> Any: + held[_name] = pass_lock_held(path) + return _real(path, *args, **kwargs) + + monkeypatch.setattr(converter, name, recording) + + assert not pass_lock_held(db_path) + process_projects_hierarchy(projects, silent=True) + + assert held == { + "trim_oversized_wal": True, + "catch_up_search_index": True, + "checkpoint_wal": True, + } + assert not pass_lock_held(db_path), "the lock outlived the pass"