diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index e946baad..77bc6eb1 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -646,14 +646,20 @@ The tool implements a SQLite-based caching system for performance: (`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 - project's refresh commits as one transaction (`CacheManager.atomic_refresh`): - while it parses, its entry rows go to a spool file beside the cache - (`.spool--`, `refresh_spool.py`), and one merge at the end - commits them with the session rows, aggregates and sidecar derived from - them, so a refresh that fails part-way leaves the cache as it was and the - next run redoes it. Only the refresh writes entry rows: a run renders what - its refresh committed, so lines written to a session while it renders - appear on the next run. A date-filtered run refreshes the + refresh of a project the cache already holds runs in one scope + (`CacheManager.atomic_refresh`): while it parses, its entry rows go to a + spool file beside the cache (`.spool--`, `refresh_spool.py`), + and a merge at the end commits them with the session rows, aggregates and + sidecar derived from them. The merge is one transaction while the rows + beyond the largest file's are at most 16 MB + (`CLAUDE_CODE_LOG_ATOMIC_MERGE_MB` overrides it), so a refresh that fails + part-way leaves the cache as it was; past that it commits per file, under a + mark (the project's message total at 0) that its last transaction lifts and + that otherwise makes the next run refresh the project in full. A cold + build, of a project with nothing cached, writes per file without a spool, + its total at 0 until it finishes. Only the refresh writes entry rows: a run + renders what its refresh committed, so lines written to a session while it + renders appear on the next run. A date-filtered run refreshes the cache as an unfiltered one would and leaves the files it wrote from its window marked stale, so the next unfiltered run re-renders them. Its refresh therefore loads whole files when they changed and the incremental diff --git a/claude_code_log/cache.py b/claude_code_log/cache.py index f71ab23d..79ae0336 100644 --- a/claude_code_log/cache.py +++ b/claude_code_log/cache.py @@ -1165,6 +1165,44 @@ def trim_oversized_wal( ) +# How a refresh's merge commits (`CacheManager._merge_refresh`). In one +# transaction, its lock hold is the sum of what each spooled file's rows +# cost to insert; per file, the longest is the largest file's. One +# transaction is kept while the difference, the rows beyond the largest +# file's, stays under this many bytes of spooled content: about half a +# second of extra hold at the rates measured on a cold merge (24-55 ms per +# spooled MB, most of it the messages indexes). A watch tick and a small +# daily delta merge at once; a refresh spread over many files, per file. +ATOMIC_MERGE_MAX_EXTRA_BYTES = 16 * 1024 * 1024 +ENV_ATOMIC_MERGE_MB = "CLAUDE_CODE_LOG_ATOMIC_MERGE_MB" + +# The page cache a merge runs with, in KiB. SQLite's default is 2 MB, and +# a merge maintains the messages indexes in random key order: on a +# 191k-row cold merge 64 MB took the copy from 6.1 s to 3.4 s. Set for the +# merge alone and given back, so it is not held by every writing +# connection (a pool of workers each holding 64 MB). +MERGE_CACHE_SIZE_KIB = 64 * 1024 + + +def atomic_merge_max_extra_bytes() -> int: + """`ATOMIC_MERGE_MAX_EXTRA_BYTES`, or ``$CLAUDE_CODE_LOG_ATOMIC_MERGE_MB``. + + The override is in megabytes and may be fractional; ``0`` merges + per file whenever more than one file is spooled, a large value + always at once. Unparsable or negative falls back to the default. + Read at each merge, as the other switches are. + """ + raw = os.environ.get(ENV_ATOMIC_MERGE_MB, "").strip() + if raw: + try: + value = float(raw) + except ValueError: + value = -1.0 + if value >= 0: + return int(value * 1024 * 1024) + return ATOMIC_MERGE_MAX_EXTRA_BYTES + + # Spools a scope could not DETACH when it ended (a statement was still # active on the connection), by connection: released at the next scope's # entry on that connection, or when the connection closes. Keyed by the @@ -1532,34 +1570,162 @@ def _drop_overlay(conn: sqlite3.Connection) -> None: def _merge_refresh( self, conn: sqlite3.Connection, pending: _PendingRefresh ) -> None: - """The one transaction an `atomic_refresh` scope ends with. + """Commit what an `atomic_refresh` scope accumulated. Runs with the spool attached and the overlay views dropped, so - the tables it writes are the tables. + the tables it writes are the tables, and with a larger page + cache (`MERGE_CACHE_SIZE_KIB`), given back after. In one + transaction when the rows beyond the largest file's are at most + `atomic_merge_max_extra_bytes()`; otherwise per file + (`_merge_per_file`), since one transaction would hold the lock + for the sum of what per file holds at most once. """ spooled = pending.spool.files() if not spooled and not pending.deferred: return + sizes = [f.content_bytes for f in spooled] + at_once = sum(sizes) - max(sizes, default=0) <= atomic_merge_max_extra_bytes() + with self._merge_page_cache(conn): + if at_once: + self._merge_at_once(conn, pending, spooled) + else: + self._merge_per_file(conn, pending, spooled) + + @staticmethod + @contextmanager + def _merge_page_cache(conn: sqlite3.Connection) -> Generator[None, None, None]: + (previous,) = conn.execute("PRAGMA cache_size").fetchone() + conn.execute(f"PRAGMA cache_size = -{MERGE_CACHE_SIZE_KIB}") + try: + yield + finally: + conn.execute(f"PRAGMA cache_size = {int(previous)}") + + @staticmethod + def _reindex(conn: sqlite3.Connection, file_ids: List[int]) -> None: + """Refresh the search index for merged files, as each direct + write does for its file (a hierarchy pass defers it to its end, + the same way).""" + from .search import ( + auto_index_enabled, + index_maintenance_deferred, + reindex_files, + ) + + if auto_index_enabled() and not index_maintenance_deferred(): + reindex_files(conn, file_ids, commit=False) + + def _merge_at_once( + self, + conn: sqlite3.Connection, + pending: _PendingRefresh, + spooled: List[SpooledFile], + ) -> None: + """The whole refresh in one transaction: rows, then derived rows.""" with write_transaction(conn): - file_ids: List[int] = [] - for spooled_file in spooled: - file_ids.append(self._merge_spooled_file(conn, spooled_file, pending)) + file_ids = [self._merge_spooled_file(conn, f, pending) for f in spooled] if file_ids: - # One index refresh for every file merged, as each - # direct write does for its file (and a hierarchy - # pass defers to its end, the same way). - from .search import ( - auto_index_enabled, - index_maintenance_deferred, - reindex_files, - ) + self._reindex(conn, file_ids) + self._update_last_updated(conn) + for write in pending.deferred: + write(conn) - if auto_index_enabled() and not index_maintenance_deferred(): - reindex_files(conn, file_ids, commit=False) + def _merge_per_file( + self, + conn: sqlite3.Connection, + pending: _PendingRefresh, + spooled: List[SpooledFile], + ) -> None: + """The refresh in one transaction per file, between two more. + + The first marks the project unfinished, its message total at 0, + which the next run reads as "refresh in full, from the cached + entries" (migration 016's heal); each file then commits its + record with its rows; the last writes the derived rows, the + aggregates among them, which lift the mark. A merge that stops + anywhere leaves every file whole and the project marked. + + The last transaction first checks that each file still carries + the record this merge wrote. Another writer may have rewritten + one in between, from a newer parse; derived rows from this + merge's parse would then read as current over those rows, so + they are not written, and the mark stays for the next run. + + With nothing deferred there are no derived rows to fall out of + step, and no mark either. + """ + if pending.deferred: + with write_transaction(conn): + self._mark_refresh_unfinished(conn) + merged: Dict[int, Optional[Tuple[Any, ...]]] = {} + for spooled_file in spooled: + file_id, record = self._merge_one_file(conn, spooled_file, pending) + merged[file_id] = record + with write_transaction(conn): + moved = [ + file_id + for file_id, record in merged.items() + if self._file_record_identity(conn, file_id) != record + ] + if moved: + logger.info( + "%d file(s) of %s rewritten by another writer during the" + " merge; left for the next run to derive", + len(moved), + self.project_path, + ) + return + if merged: self._update_last_updated(conn) for write in pending.deferred: write(conn) + def has_cached_files(self) -> bool: + """Whether the cache holds any file of this project. + + A refresh of a project with none is a cold build, which + `ensure_fresh_cache` runs without a spool. + """ + if self._project_id is None: + return False + with self._get_connection() as conn: + rows = conn.execute( + "SELECT 1 FROM cached_files WHERE project_id = ? LIMIT 1", + (self._project_id,), + ).fetchall() + return bool(rows) + + def _mark_refresh_unfinished(self, conn: sqlite3.Connection) -> None: + conn.execute( + "UPDATE projects SET total_message_count = 0 WHERE id = ?", + (self._project_id,), + ) + + def _merge_one_file( + self, + conn: sqlite3.Connection, + spooled_file: SpooledFile, + pending: _PendingRefresh, + ) -> Tuple[int, Optional[Tuple[Any, ...]]]: + """One spooled file in a transaction of its own; returns its id + and the record it was left with.""" + with write_transaction(conn): + file_id = self._merge_spooled_file(conn, spooled_file, pending) + self._reindex(conn, [file_id]) + return file_id, self._file_record_identity(conn, file_id) + + @staticmethod + def _file_record_identity( + conn: sqlite3.Connection, file_id: int + ) -> Optional[Tuple[Any, ...]]: + """What tells one write of a file's record from another.""" + row = conn.execute( + "SELECT source_mtime, source_size, cached_mtime, message_count" + " FROM cached_files WHERE id = ?", + (file_id,), + ).fetchone() + return None if row is None else tuple(row) + def _merge_spooled_file( self, conn: sqlite3.Connection, diff --git a/claude_code_log/converter.py b/claude_code_log/converter.py index dbd26598..489c0bbf 100644 --- a/claude_code_log/converter.py +++ b/claude_code_log/converter.py @@ -4896,13 +4896,29 @@ def ensure_fresh_cache_detailed( if not needs_update: return CacheRefresh.NONE # Cache is already fresh - # One transaction for the whole refresh (`atomic_refresh`), - # whichever path it takes: the per-file entry writes go to a - # spool, the session rows, sidecar and aggregates derived from - # them are deferred, and the scope's end commits all of it at - # once. A refresh that stops anywhere in between leaves the - # cache as it was, with nothing recorded as current. - with cache_manager.atomic_refresh(): + # Over rows the cache already holds, the refresh is one scope + # (`atomic_refresh`), whichever path it takes: the per-file + # entry writes go to a spool, the session rows, sidecar and + # aggregates derived from them are deferred, and the scope's + # end merges all of it, in one transaction or, when the rows + # are spread over many files, one per file under a mark that + # the next run heals. A refresh that stops anywhere never + # leaves a file recorded as current over derived rows that + # predate it. + # + # A cold build (no cached file for the project) needs no + # spool: there are no derived rows to fall out of step with, + # and the project's total stays 0 until the aggregates, its + # last write, so a build that stops reads as unfinished and the + # next run completes it. Its files commit one by one as they + # are parsed, which keeps the lock hold of a large project's + # first build to its largest file, and its disk to the cache. + scope = ( + cache_manager.atomic_refresh() + if cache_manager.has_cached_files() + else contextlib.nullcontext() + ) + with scope: # Streaming stage 4: when the update is driven purely by # changed files over a populated cache, refresh from those # files' bounded closure instead of loading the whole diff --git a/claude_code_log/refresh_spool.py b/claude_code_log/refresh_spool.py index 64c7b030..53af2d84 100644 --- a/claude_code_log/refresh_spool.py +++ b/claude_code_log/refresh_spool.py @@ -73,7 +73,8 @@ cached_mtime REAL NOT NULL, message_count INTEGER NOT NULL, subagents_fingerprint TEXT, - content_version TEXT + content_version TEXT, + content_bytes INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE messages ( id INTEGER PRIMARY KEY, @@ -176,7 +177,8 @@ class SpooledFile: or ``append`` (the rows are a tail, to follow exactly ``expected_existing`` rows already in the cache). The remaining fields are the ``cached_files`` row as of the parse, which the merge - writes verbatim. + writes verbatim, except ``content_bytes``: the size of the spooled + rows' content, which decides how the merge commits. """ id: int @@ -190,6 +192,7 @@ class SpooledFile: message_count: int subagents_fingerprint: Optional[str] content_version: Optional[str] + content_bytes: int class RefreshSpool: @@ -257,8 +260,17 @@ def add( {**record, "mode": mode, "expected_existing": expected_existing}, ) file_id = cursor.lastrowid or 0 - self._conn.executemany( - _INSERT_ROW, ({**row, "file_id": file_id} for row in rows) + size = 0 + + def counted() -> Iterable[Dict[str, Any]]: + nonlocal size + for row in rows: + size += len(row["content"]) + yield {**row, "file_id": file_id} + + self._conn.executemany(_INSERT_ROW, counted()) + self._conn.execute( + "UPDATE files SET content_bytes = ? WHERE id = ?", (size, file_id) ) self._conn.execute("COMMIT") except BaseException: @@ -271,7 +283,7 @@ def files(self) -> List[SpooledFile]: rows = self._conn.execute( """SELECT id, file_name, file_path, mode, expected_existing, source_mtime, source_size, cached_mtime, message_count, - subagents_fingerprint, content_version + subagents_fingerprint, content_version, content_bytes FROM files ORDER BY id""" ).fetchall() return [SpooledFile(*row) for row in rows] diff --git a/dev-docs/application_model.md b/dev-docs/application_model.md index 7dbda4ca..3ed85571 100644 --- a/dev-docs/application_model.md +++ b/dev-docs/application_model.md @@ -365,19 +365,21 @@ left the files current and the session rows as old as the renders, and every later run reported the project as cached while its renders lacked the new messages; migration 015's `refresh_pending` flag carried such a failure to the next run, with a limit (two processes refreshing one -project could clear each other's) and special rules of its own. Now -`ensure_fresh_cache_detailed` runs either path inside -`CacheManager.atomic_refresh()`: while the refresh parses, each file's +project could clear each other's) and special rules of its own. Now, +over a project the cache already holds, `ensure_fresh_cache_detailed` +runs either path inside `CacheManager.atomic_refresh()`: while the +refresh parses, each file's rows go to a private **spool** file beside the cache (`.spool--`, [`refresh_spool.py`](../claude_code_log/refresh_spool.py); journal in memory, nothing synced, this process's own), with no shared lock held and the CPU work of serialising where it was; the derived writes are -deferred; and one write transaction at the scope's end attaches the -spool, copies each file's rows in with `INSERT … SELECT` (no Python per -row under the lock, and less time than the direct `executemany` took), -runs the deferred writes, and commits. A refresh that stops anywhere -leaves the cache as it was and the spool is removed; the next run, or +deferred; and the scope's end attaches the spool, copies each file's +rows in with `INSERT … SELECT` (no Python per row under the lock, and +less time than the direct `executemany` took), runs the deferred writes, +and commits, in one write transaction or one per file (below). A +refresh whose single transaction stops leaves the cache as it was and +the spool is removed; the next run, or the pass's in-pass retry, redoes it, re-parsing the modified files. Only a process killed mid-parse leaves a spool behind, and a pass removes the spools of dead processes before it starts, under its lock. A @@ -399,6 +401,36 @@ one-time heal: a project it still marked gets its message total zeroed, which the next run reads as a cache to refresh in full, from the cached entries. +**How the merge commits, and the cold build.** One transaction holds +the write lock for the sum of what each file's rows cost to insert, most +of it the ten `messages` indexes maintained in random key order (a cold +merge of 191k rows: 0.84 s to copy, about 6 s with the indexes, 1 s to +commit); per file, the longest hold is the largest file's. So +`_merge_refresh` merges in one transaction while the spooled content +beyond the largest file's is at most `ATOMIC_MERGE_MAX_EXTRA_BYTES` +(16 MB, about half a second of extra hold; `CLAUDE_CODE_LOG_ATOMIC_MERGE_MB` +overrides it, `0` meaning per file whenever more than one file is +spooled), which keeps a watch tick and a small daily delta atomic at no +cost. Past it, `_merge_per_file` commits per file between two more +transactions: the first marks the project unfinished by zeroing its +message total (016's heal: the next run refreshes it in full, from the +cached entries), and the last writes the derived rows, whose aggregates +lift the mark, but only if every file it merged still carries the +record it wrote: a file another writer rewrote in between, from a newer +parse, would otherwise read as current under derived rows from the +older one, so the mark stays and the next run derives from what the +cache holds. Either way the merge runs with a 64 MB page cache +(`MERGE_CACHE_SIZE_KIB`; SQLite's default is 2 MB, which the indexes +thrash: 6.1 s against 3.4 s for the copy above), set for the merge and +given back, so no long-lived writing connection keeps it. A **cold +build**, of a project with no cached file (a new project, or one whose +rows a version change discarded), opens no scope at all: there are no +derived rows to fall out of step with, and the project's total is 0 +until the aggregates, the refresh's last write, so a build that stops +reads as unfinished and the next run completes it. Its files commit one +by one as they are parsed, which keeps a large project's first build to +its largest file's hold and adds no spool to its disk. + **A run renders what its refresh committed.** Only the refresh writes entry rows; the render reads them and writes none. A session file can grow between the two, since sessions are written to while a run converts diff --git a/test/test_cache_merge_modes.py b/test/test_cache_merge_modes.py new file mode 100644 index 00000000..20bf42ec --- /dev/null +++ b/test/test_cache_merge_modes.py @@ -0,0 +1,377 @@ +"""How a refresh commits: one transaction, one per file, or no spool at all. + +A refresh over a project the cache already holds spools its rows and +merges them at its end. In one transaction when that costs little more +lock than merging file by file (the rows beyond the largest file's are +at most `atomic_merge_max_extra_bytes`); otherwise one transaction per +file, the first marking the project unfinished (its message total at +0, which the next run reads as "refresh in full") and the last writing +the derived rows, which restore it, only if every file it merged is +still as it left it. A cold build, of a project with no cached rows, +spools nothing and writes per file as it parses. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path +from typing import Any, List + +import pytest + +from claude_code_log import cache as cache_mod +from claude_code_log import converter +from claude_code_log.cache import ( + ATOMIC_MERGE_MAX_EXTRA_BYTES, + MERGE_CACHE_SIZE_KIB, + CacheManager, + atomic_merge_max_extra_bytes, + get_library_version, +) +from test.test_cache_atomic_refresh import ( + _append, + _dump, + _entries, + _grow, + _locked, + _manager, + _never_busy, + _pass, + _refresh, + _rows_by_file, + _spools, + _turn, + _fail_second_file_write, + after_columns, +) + + +@pytest.fixture +def project(tmp_path: Path) -> Path: + """Two sessions of one turn each (as in test_cache_atomic_refresh).""" + project = tmp_path / "projects" / "-proj" + project.mkdir(parents=True) + for sid in ("sess-alpha", "sess-beta"): + _append(project / f"{sid}.jsonl", _turn(sid, 0, f"first {sid}")) + return project + + +def _per_file(monkeypatch: pytest.MonkeyPatch) -> None: + """Every merge with more than one file goes per file.""" + monkeypatch.setenv("CLAUDE_CODE_LOG_ATOMIC_MERGE_MB", "0") + + +def _total(db_path: Path) -> int: + conn = sqlite3.connect(db_path) + try: + return conn.execute("SELECT total_message_count FROM projects").fetchone()[0] + finally: + conn.close() + + +def _sessions(project: Path) -> dict[str, int]: + data = _manager(project).get_cached_project_data() + assert data is not None + return {sid: s.message_count for sid, s in data.sessions.items()} + + +def _built(project: Path) -> CacheManager: + """A manager over a cache that already holds the project.""" + manager = _manager(project) + files = sorted(project.glob("*.jsonl")) + _refresh(manager, files, files) + return manager + + +def _begins(manager: CacheManager, files: List[Path], changed: List[Path]) -> List[str]: + """The BEGINs on the scope's connection while it refreshes ``changed``.""" + begins: List[str] = [] + with manager.atomic_refresh(): + conn = manager._shared_conn # pyright: ignore[reportPrivateUsage] + assert conn is not None + conn.set_trace_callback( + lambda sql: begins.append(sql) if sql.upper().startswith("BEGIN") else None + ) + _refresh(manager, files, changed) + return begins + + +# ---- the threshold ------------------------------------------------------- + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (None, ATOMIC_MERGE_MAX_EXTRA_BYTES), + ("", ATOMIC_MERGE_MAX_EXTRA_BYTES), + ("0", 0), + ("0.5", 512 * 1024), + ("64", 64 * 1024 * 1024), + ("-1", ATOMIC_MERGE_MAX_EXTRA_BYTES), + ("lots", ATOMIC_MERGE_MAX_EXTRA_BYTES), + ], +) +def test_the_threshold_reads_its_override_in_megabytes( + monkeypatch: pytest.MonkeyPatch, value: str | None, expected: int +) -> None: + if value is None: + monkeypatch.delenv("CLAUDE_CODE_LOG_ATOMIC_MERGE_MB", raising=False) + else: + monkeypatch.setenv("CLAUDE_CODE_LOG_ATOMIC_MERGE_MB", value) + assert atomic_merge_max_extra_bytes() == expected + assert ATOMIC_MERGE_MAX_EXTRA_BYTES == 16 * 1024 * 1024 + + +# ---- which mode ------------------------------------------------------------ + + +def test_a_small_refresh_merges_in_one_transaction( + project: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.delenv("CLAUDE_CODE_LOG_ATOMIC_MERGE_MB", raising=False) + manager = _built(project) + files, changed = _grow(project) + assert len(_begins(manager, files, changed)) == 1 + + +def test_a_spread_refresh_merges_per_file( + project: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The mark, one transaction per spooled file, the derived rows.""" + _per_file(monkeypatch) + manager = _built(project) + files, changed = _grow(project) + assert len(_begins(manager, files, changed)) == len(changed) + 2 + assert _sessions(project) == {"sess-alpha": 4, "sess-beta": 2, "sess-gamma": 2} + assert _total(manager.db_path) == 8 + + +def test_a_refresh_of_one_file_stays_atomic_whatever_its_size( + project: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Merging per file cannot shorten a one-file merge's hold.""" + _per_file(monkeypatch) + manager = _built(project) + alpha = project / "sess-alpha.jsonl" + _append(alpha, _turn("sess-alpha", 1, "late alpha")) + files = sorted(project.glob("*.jsonl")) + assert len(_begins(manager, files, [alpha])) == 1 + + +def test_a_cold_build_writes_per_file_without_a_spool( + project: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + spools: List[object] = [] + from claude_code_log.refresh_spool import RefreshSpool + + class Recording(RefreshSpool): + def __init__(self, db_path: Path) -> None: + spools.append(db_path) + super().__init__(db_path) + + monkeypatch.setattr(cache_mod, "RefreshSpool", Recording) + manager = _manager(project) + converter.ensure_fresh_cache(project, manager, silent=True) + assert spools == [] + assert _sessions(project) == {"sess-alpha": 2, "sess-beta": 2} + assert _total(manager.db_path) == 4 + + # Positive control: over the rows it now holds, a refresh spools. + _grow(project) + converter.ensure_fresh_cache(project, _manager(project), silent=True) + assert len(spools) == 1 + assert _total(manager.db_path) == 8 + + +@pytest.mark.parametrize("mode", ["one", "per-file"]) +def test_both_modes_leave_the_same_cache( + project: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, mode: str +) -> None: + """Row for row, per file is the one-transaction merge.""" + if mode == "per-file": + _per_file(monkeypatch) + manager = _built(project) + files, changed = _grow(project) + with manager.atomic_refresh(): + _refresh(manager, files, changed) + + monkeypatch.delenv("CLAUDE_CODE_LOG_ATOMIC_MERGE_MB", raising=False) + direct = CacheManager(project, get_library_version(), db_path=tmp_path / "d.db") + _refresh(direct, files, files) + + assert _rows_by_file(manager.db_path) == _rows_by_file(direct.db_path) + merged, reference = _dump(manager.db_path), _dump(direct.db_path) + assert [r[2:] for r in merged["sessions"]] == [r[2:] for r in reference["sessions"]] + cols = after_columns(manager.db_path) + skip = {"id", "cache_created", "last_updated"} + assert [v for c, v in zip(cols, merged["projects"][0]) if c not in skip] == [ + v for c, v in zip(cols, reference["projects"][0]) if c not in skip + ] + + +def test_the_merge_runs_with_a_larger_page_cache_and_gives_it_back( + project: Path, +) -> None: + manager = _built(project) + files, changed = _grow(project) + during: List[int] = [] + real = CacheManager._merge_spooled_file # pyright: ignore[reportPrivateUsage] + + def spy(self: CacheManager, conn: sqlite3.Connection, *args: Any) -> int: + during.append(conn.execute("PRAGMA cache_size").fetchone()[0]) + return real(self, conn, *args) + + with manager.batch(): + conn = manager._shared_conn # pyright: ignore[reportPrivateUsage] + assert conn is not None + before = conn.execute("PRAGMA cache_size").fetchone()[0] + with pytest.MonkeyPatch.context() as m: + m.setattr(CacheManager, "_merge_spooled_file", spy) + with manager.atomic_refresh(): + _refresh(manager, files, changed) + after = conn.execute("PRAGMA cache_size").fetchone()[0] + assert during and set(during) == {-MERGE_CACHE_SIZE_KIB} + assert after == before + + +# ---- interrupted, overtaken ------------------------------------------------ + + +def _fail_mark(m: pytest.MonkeyPatch) -> None: + m.setattr(CacheManager, "_mark_refresh_unfinished", _locked) + + +def _fail_second_file(m: pytest.MonkeyPatch) -> None: + real = CacheManager._merge_spooled_file # pyright: ignore[reportPrivateUsage] + calls: List[str] = [] + + def second_fails(self: CacheManager, conn: sqlite3.Connection, *args: Any) -> int: + calls.append("x") + if len(calls) == 2: + _locked() + return real(self, conn, *args) + + m.setattr(CacheManager, "_merge_spooled_file", second_fails) + + +def _fail_derived(m: pytest.MonkeyPatch) -> None: + def deferred_failure(self: CacheManager, **_kwargs: Any) -> None: + self._write_or_defer(_locked) # pyright: ignore[reportPrivateUsage] + + m.setattr(CacheManager, "update_project_aggregates", deferred_failure) + + +@pytest.mark.parametrize( + "inject", + [_fail_mark, _fail_second_file, _fail_derived], + ids=["mark", "file", "derived"], +) +def test_an_interrupted_per_file_merge_is_marked_and_heals( + project: Path, monkeypatch: pytest.MonkeyPatch, inject: Any +) -> None: + _per_file(monkeypatch) + manager = _built(project) + before = _dump(manager.db_path) + files, changed = _grow(project) + + with monkeypatch.context() as m: + inject(m) + with pytest.raises(sqlite3.OperationalError, match="locked"): + with manager.atomic_refresh(): + _refresh(manager, files, changed) + assert _spools(manager.db_path) == [] + if inject is _fail_mark: + # Nothing committed: the cache is as it was, total included. + assert _dump(manager.db_path) == before + else: + assert _total(manager.db_path) == 0 + + converter.ensure_fresh_cache(project, _manager(project), silent=True) + assert _sessions(project) == {"sess-alpha": 4, "sess-beta": 2, "sess-gamma": 2} + assert _total(manager.db_path) == 8 + + +def test_a_per_file_merge_overtaken_by_another_writer_stays_marked( + project: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Between A's file transactions, B rewrites a file A has merged, + from a newer parse. A's derived rows describe A's parse: written, + they would read current over B's rows. A leaves the mark instead, + and the next run derives from what the cache holds.""" + _per_file(monkeypatch) + manager = _built(project) + files, changed = _grow(project) + alpha = project / "sess-alpha.jsonl" + overtaken: List[str] = [] + real = CacheManager._merge_one_file # pyright: ignore[reportPrivateUsage] + + def then_b( + self: CacheManager, conn: sqlite3.Connection, spooled: Any, *args: Any + ) -> Any: + result = real(self, conn, spooled, *args) + if not overtaken and spooled.file_name == alpha.name: + overtaken.append(spooled.file_name) + _append(alpha, _turn("sess-alpha", 2, "turn two")) + b = CacheManager(project, get_library_version(), db_path=manager.db_path) + b.save_cached_entries(alpha, _entries(alpha)) + return result + + with monkeypatch.context() as m: + m.setattr(CacheManager, "_merge_one_file", then_b) + with manager.atomic_refresh(): + _refresh(manager, files, changed) + assert overtaken == [alpha.name] # positive control: B ran in between + assert len(_rows_by_file(manager.db_path)["sess-alpha.jsonl"]) == 6 + assert _total(manager.db_path) == 0 + + converter.ensure_fresh_cache(project, _manager(project), silent=True) + assert _sessions(project) == {"sess-alpha": 6, "sess-beta": 2, "sess-gamma": 2} + assert _total(manager.db_path) == 10 + + +def test_an_interrupted_cold_build_is_marked_and_heals( + project: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + projects_dir = project.parent + db_path = projects_dir / "cache.db" + monkeypatch.setenv("CLAUDE_CODE_LOG_CACHE_PATH", str(db_path)) + with monkeypatch.context() as m: + _fail_second_file_write(m) + m.setattr(converter, "is_busy_database_error", _never_busy) + _pass(projects_dir) + assert "projects failed" in capsys.readouterr().out + assert _spools(db_path) == [] + # Per file: the first file is in, the project reads unfinished. + assert len(_rows_by_file(db_path)) == 1 + assert _total(db_path) == 0 + + _pass(projects_dir) + assert _sessions(project) == {"sess-alpha": 2, "sess-beta": 2} + assert _total(db_path) == 4 + + +def test_a_per_file_merge_with_nothing_derived_takes_no_mark( + project: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A scope that writes rows only (as `atomic_refresh` allows any + caller): nothing derived can fall out of step with them, and nothing + would lift a mark, so the project keeps its total instead of being + refreshed in full by the next run for no reason.""" + _per_file(monkeypatch) + manager = _built(project) + files, changed = _grow(project) + begins: List[str] = [] + with manager.atomic_refresh(): + conn = manager._shared_conn # pyright: ignore[reportPrivateUsage] + assert conn is not None + conn.set_trace_callback( + lambda sql: begins.append(sql) if sql.upper().startswith("BEGIN") else None + ) + for f in changed: + manager.save_cached_entries(f, _entries(f)) + # Per file (one per file, then the last), and no mark before them. + assert len(begins) == len(changed) + 1 + assert len(_rows_by_file(manager.db_path)["sess-gamma.jsonl"]) == 2 + assert _total(manager.db_path) == 4