From 11d65606d18885c090de9c707d112afa2142541f Mon Sep 17 00:00:00 2001 From: Christian Boos Date: Sat, 7 Nov 2026 16:16:50 +0100 Subject: [PATCH 1/2] Merge a spread refresh per file, and build a cold project without a spool One merge transaction holds the write lock for the sum of what each spooled file's rows cost to insert, mostly the messages indexes: on a cold build of a 191k-row project that was one 7-8 s hold and a WAL the size of the project. Merging per file holds it at most for the largest file's rows. The merge now commits in one transaction while the spooled content beyond the largest file's is at most 16 MB (ATOMIC_MERGE_MAX_EXTRA_BYTES, CLAUDE_CODE_LOG_ATOMIC_MERGE_MB to override), so a watch tick and a small delta stay atomic. Past that it commits per file between two more transactions: the first zeroes the project's message total, the mark migration 016 already heals by refreshing in full; the last writes the derived rows, which lift it, only if every merged file still has the record the merge wrote, since a file another writer rewrote in between would otherwise read as current under older derived rows. The spool records each file's content bytes to decide. Either way the merge runs with a 64 MB page cache, given back after it. A cold build, of a project with no cached file, opens no scope: there are no derived rows to fall out of step with, and the total stays 0 until the aggregates, its last write. Its files commit as they are parsed, without a spool. Co-Authored-By: Claude Opus 5.5 --- CONTRIBUTING.md | 22 +- claude_code_log/cache.py | 196 +++++++++++++++-- claude_code_log/converter.py | 30 ++- claude_code_log/refresh_spool.py | 22 +- dev-docs/application_model.md | 48 ++++- test/test_cache_merge_modes.py | 352 +++++++++++++++++++++++++++++++ 6 files changed, 627 insertions(+), 43 deletions(-) create mode 100644 test/test_cache_merge_modes.py 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..08605eb9 --- /dev/null +++ b/test/test_cache_merge_modes.py @@ -0,0 +1,352 @@ +"""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 From 3c178db3797017233e67c4103fc650b71ba10719 Mon Sep 17 00:00:00 2001 From: Christian Boos Date: Sat, 7 Nov 2026 16:31:53 +0100 Subject: [PATCH 2/2] Pin that a per-file merge with nothing derived takes no mark A scope may write rows only. With nothing deferred, nothing derived can fall out of step with them and nothing would lift the mark, so the per-file merge skips it; marking anyway would leave the project at 0 and send the next run into a full refresh for no reason. Nothing tested that branch: taking the mark unconditionally left every test green. Co-Authored-By: Claude Opus 5.5 --- test/test_cache_merge_modes.py | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/test/test_cache_merge_modes.py b/test/test_cache_merge_modes.py index 08605eb9..20bf42ec 100644 --- a/test/test_cache_merge_modes.py +++ b/test/test_cache_merge_modes.py @@ -350,3 +350,28 @@ def test_an_interrupted_cold_build_is_marked_and_heals( _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