diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 77bc6eb1..ab121958 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -657,7 +657,10 @@ The tool implements a SQLite-based caching system for performance: 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 + its total at 0 until it finishes. Either reads the project's file records + as it opens and writes the derived rows only if no other refresh has + rewritten a file since; otherwise it leaves the total at 0 for the next + run. 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 diff --git a/claude_code_log/cache.py b/claude_code_log/cache.py index 79ae0336..3566c77e 100644 --- a/claude_code_log/cache.py +++ b/claude_code_log/cache.py @@ -1221,6 +1221,11 @@ def _forget_leftover_spool(conn: sqlite3.Connection) -> None: pass +# A file's record identity (`CacheManager._file_record_identity`), by +# file name, for the files of one project. +_FileRecords = Dict[str, Tuple[Any, ...]] + + class _PendingRefresh: """What an `atomic_refresh` scope accumulates until its merge. @@ -1232,16 +1237,32 @@ class _PendingRefresh: rows are no longer what the tail expected to follow, it rewrites the file whole from those entries, as the direct path's refused append falls back to a full save. + + ``records`` is the project's file records as the scope opened + (`CacheManager._project_file_records`): the merge writes the derived + rows only if every file it did not write itself still carries its + own (`CacheManager._moved_files`). """ - def __init__(self, spool: RefreshSpool) -> None: + def __init__(self, spool: RefreshSpool, records: _FileRecords) -> None: self.spool = spool + self.records = records self.deferred: List[Callable[[sqlite3.Connection], None]] = [] self.appends: Dict[ str, Tuple[Path, List[TranscriptEntry], Optional[str], os.stat_result] ] = {} +class _ColdBuild: + """What a `CacheManager.cold_build` scope remembers until its + aggregates: the project's file records as it opened, and the + record of each file it has written since.""" + + def __init__(self, records: _FileRecords) -> None: + self.records = records + self.written: Dict[str, Optional[Tuple[Any, ...]]] = {} + + class CacheManager: """SQLite-based cache manager for Claude Code Log.""" @@ -1289,6 +1310,7 @@ def __init__( # Inside an `atomic_refresh` scope: the spool and the deferred # derived writes; None means every write lands at once. self._refresh: Optional[_PendingRefresh] = None + self._cold: Optional[_ColdBuild] = None self._project_id: Optional[int] = None if read_only: @@ -1417,10 +1439,29 @@ def atomic_refresh(self) -> Generator[None, None, None]: So a refresh that stops at any point (a lock-out, a kill, a full disk) never records a file as current over session rows that predate its entries, and a reader on another connection - sees either the whole refresh or none of it. Two refreshes of - one project overlapping commit two self-consistent snapshots: - the later wins, and if it parsed older files its - ``cached_files`` rows say so. + sees either the whole refresh or none of it. + + Another refresh of the project may commit while this one runs + (only `process_projects_hierarchy` holds the pass lock). A file + this one parsed is written with the record of that parse, so if + the other's was newer the file reads modified and the next run + reparses it. A file it took from the cache is not written, and + its record would read current over derived rows computed from + the rows it replaced: so the scope reads every file record of + the project as it opens, before the body's first read, and the + merge writes the derived rows only if each file it did not + write still carries that record (`_moved_files`). If one moved, + the merge commits the rows and leaves the project marked + unfinished (its total at 0), and the next run refreshes it in + full from what the cache then holds. + + The snapshot comes after the caller's ``get_modified_files``, + which is only a gate: a commit in between is one the body's + reads already see, so the check firing on it costs a full + refresh and nothing else. Files the body parses come from disk + or from an ``entry_store`` hit, which is stamped with the source + file's size and mtime (`entry_store.stamp_file`), so never older + than the disk, nor than a record another writer took from it. Needs a single connection for the scope, so it opens a ``batch()`` of its own when the caller has none. Nesting joins @@ -1445,7 +1486,8 @@ def atomic_refresh(self) -> Generator[None, None, None]: "transaction; a write on it bypassed write_transaction" ) self._release_leftover_spool(conn) - pending = _PendingRefresh(RefreshSpool(self.db_path)) + records = self._project_file_records(conn) + pending = _PendingRefresh(RefreshSpool(self.db_path), records) try: conn.execute("ATTACH DATABASE ? AS spool", (str(pending.spool.path),)) try: @@ -1627,6 +1669,16 @@ def _merge_at_once( if file_ids: self._reindex(conn, file_ids) self._update_last_updated(conn) + if not pending.deferred: + return + # The spooled files' records are written above, in this + # transaction: only the others can have moved. + own: Dict[str, Optional[Tuple[Any, ...]]] = { + f.file_name: None for f in spooled + } + if self._moved_files(conn, pending.records, own): + self._mark_refresh_unfinished(conn) + return for write in pending.deferred: write(conn) @@ -1646,10 +1698,12 @@ def _merge_per_file( 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. + the record this merge wrote, and every other file the record it + had as the scope opened (`_moved_files`). Another writer may + have rewritten one in between, from a newer parse; derived rows + from this merge's view 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. @@ -1657,23 +1711,12 @@ def _merge_per_file( if pending.deferred: with write_transaction(conn): self._mark_refresh_unfinished(conn) - merged: Dict[int, Optional[Tuple[Any, ...]]] = {} + merged: Dict[str, Optional[Tuple[Any, ...]]] = {} for spooled_file in spooled: - file_id, record = self._merge_one_file(conn, spooled_file, pending) - merged[file_id] = record + _file_id, record = self._merge_one_file(conn, spooled_file, pending) + merged[spooled_file.file_name] = 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, - ) + if pending.deferred and self._moved_files(conn, pending.records, merged): return if merged: self._update_last_updated(conn) @@ -1695,6 +1738,82 @@ def has_cached_files(self) -> bool: ).fetchall() return bool(rows) + @contextmanager + def cold_build(self) -> Generator[None, None, None]: + """The scope of a refresh without a spool: a project's first build. + + Its files commit one by one as they are parsed, and the project's + total stays 0 until the aggregates, its last write. Another + refresh may commit in between (over the files already written, + so in an `atomic_refresh` scope of its own), and the session + rows this build derives would then describe an older view of a + file whose record reads current. So the scope reads the + project's file records as it opens, remembers the record it + leaves on each file it writes, and the aggregates are written + only if every file still carries the one or the other + (`_moved_files`); otherwise the total is written 0 again, and + the next run refreshes the project in full. + + Nesting, a read-only instance or no project row: no scope. + """ + if ( + self._cold is not None + or self._refresh is not None + or self._read_only + or self._project_id is None + ): + yield + return + with self.batch(), self._get_connection() as conn: + self._cold = _ColdBuild(self._project_file_records(conn)) + try: + yield + finally: + self._cold = None + + def _project_file_records(self, conn: sqlite3.Connection) -> _FileRecords: + """Every file record of the project, by name, from the table + itself (not a scope's overlay).""" + rows = conn.execute( + "SELECT file_name, source_mtime, source_size, cached_mtime," + " message_count FROM main.cached_files WHERE project_id = ?", + (self._project_id,), + ).fetchall() + return {row[0]: tuple(row[1:]) for row in rows} + + def _moved_files( + self, + conn: sqlite3.Connection, + records: _FileRecords, + own: Dict[str, Optional[Tuple[Any, ...]]], + ) -> List[str]: + """The files whose record another writer has changed since a + refresh read ``records``, under the caller's write lock. + + ``own`` holds the files the refresh wrote itself, with the + record it left (None: written in this transaction, so it cannot + have moved); every other file must still carry its record in + ``records``, and a file on one side only has moved. + """ + current = self._project_file_records(conn) + moved = sorted( + name + for name in set(current) | set(records) | set(own) + if ( + own[name] is not None and current.get(name) != own[name] + if name in own + else current.get(name) != records.get(name) + ) + ) + if moved: + logger.info( + "%d file(s) of %s rewritten by another writer during the" + " refresh; left for the next run to derive", + len(moved), + self.project_path, + ) + return moved + def _mark_refresh_unfinished(self, conn: sqlite3.Connection) -> None: conn.execute( "UPDATE projects SET total_message_count = 0 WHERE id = ?", @@ -2258,6 +2377,10 @@ def save_cached_entries( file_id = self._replace_file_rows( conn, jsonl_path, serialized_entries, source_stat, subagents_fp ) + if self._cold is not None: + self._cold.written[jsonl_path.name] = self._file_record_identity( + conn, file_id + ) # Keep the full-archive search index in step with the rows we # just replaced, inside the same transaction. This is a no-op @@ -2418,7 +2541,7 @@ def extend_cached_entries( # refused had it seen them. A refusal writes nothing, so the # commit it ends with is empty. Inside a `batch()` scope already # in a transaction, that one covers it. - return self._append_under_lock( + appended_ok = self._append_under_lock( conn, jsonl_path, all_entries, @@ -2427,6 +2550,11 @@ def extend_cached_entries( source_stat, subagents_fp, ) + if appended_ok and self._cold is not None: + self._cold.written[jsonl_path.name] = self._file_record_identity( + conn, self._file_id_under_lock(conn, jsonl_path.name) + ) + return appended_ok def _append_under_lock( self, @@ -2600,6 +2728,20 @@ def write(conn: sqlite3.Connection) -> None: ), ) + cold = self._cold + if cold is not None and self._refresh is None: + # A cold build's last write: if another writer moved a file + # since the build read or wrote it, the session rows already + # written may describe an older view; leave the project + # unfinished for the next run instead (`cold_build`). + def guarded(conn: sqlite3.Connection) -> None: + if self._moved_files(conn, cold.records, cold.written): + self._mark_refresh_unfinished(conn) + else: + write(conn) + + self._write_or_defer(guarded) + return self._write_or_defer(write) def save_session_sidecar(self, sidecar: SessionSidecar) -> None: diff --git a/claude_code_log/converter.py b/claude_code_log/converter.py index 489c0bbf..93165f6c 100644 --- a/claude_code_log/converter.py +++ b/claude_code_log/converter.py @@ -4913,10 +4913,15 @@ def ensure_fresh_cache_detailed( # 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. + # + # Either scope reads the project's file records as it opens, + # after the gate above and before any read the derived rows + # come from, and writes them only if no other refresh has moved + # a file since (`CacheManager._moved_files`). scope = ( cache_manager.atomic_refresh() if cache_manager.has_cached_files() - else contextlib.nullcontext() + else cache_manager.cold_build() ) with scope: # Streaming stage 4: when the update is driven purely by diff --git a/dev-docs/application_model.md b/dev-docs/application_model.md index 3ed85571..8e3a2fb9 100644 --- a/dev-docs/application_model.md +++ b/dev-docs/application_model.md @@ -384,9 +384,7 @@ 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 reader on another connection sees either the whole refresh or none of -it, and two refreshes of one project overlapping commit two -self-consistent snapshots: the later wins, and if it parsed older files -its `cached_files` rows say so and the next run repairs it. Inside the +it. Inside the scope, two TEMP views named `cached_files` and `messages` shadow the tables for every read on the scope's connection (the table minus the rows the spooled files replace, `UNION ALL` the spool's rows), because @@ -415,21 +413,45 @@ 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 +lift the mark, but only if no file moved (below). 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. +rows a version change discarded), takes no spool (`CacheManager.cold_build`): +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. + +**Two refreshes of one project overlapping.** Only a hierarchy pass +holds the pass lock; a single-project convert, `watch `, a +single-session file or the TUI's export refresh beside it. A file a +refresh parsed it writes with the record of that parse, so if another +refresh committed a newer one in between, the file reads modified and +the next run reparses it. A file it took from the cache it does not +write, and if the other rewrote that one before this refresh's derived +rows landed, those rows (an incremental refresh's total included, which +it computes as a delta over the total it read) would describe the rows +the other replaced, under a record that reads current, and nothing +would recompute them. So both scopes read every file record of the +project as they open, after the modified-file gate (which only decides +whether to refresh: a commit before the snapshot is one the scope's +reads see) and before any read the derived rows come from, and write +the derived rows only if every file still carries its record: the one +the refresh left on the files it wrote (none to compare in the +one-transaction merge, which writes them in the same transaction), the +one in the snapshot on every other, a file on one side only counting as +moved (`_moved_files`). Otherwise the derived rows are not written and +the project's total is left at 0, so the next run refreshes it in full +from what the cache then holds; the one-transaction merge still commits +the rows it spooled. A commit after the snapshot that the refresh's +reads did see trips the check as well; it costs that one full refresh. A +file parsed from an `entry_store` hit is no exception: the store stamps +its entries with the source file's size and mtime, so a hit is never +older than the 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 diff --git a/test/test_cache_merge_modes.py b/test/test_cache_merge_modes.py index 20bf42ec..b79fc357 100644 --- a/test/test_cache_merge_modes.py +++ b/test/test_cache_merge_modes.py @@ -9,11 +9,17 @@ 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. + +Another refresh can commit while one runs. In every mode, the derived +rows are written only if each file the refresh did not write still +carries the record it had as the refresh began; otherwise the project +stays marked for the next run. """ from __future__ import annotations import sqlite3 +import threading from pathlib import Path from typing import Any, List @@ -26,6 +32,7 @@ MERGE_CACHE_SIZE_KIB, CacheManager, atomic_merge_max_extra_bytes, + connection_lease, get_library_version, ) from test.test_cache_atomic_refresh import ( @@ -375,3 +382,208 @@ def test_a_per_file_merge_with_nothing_derived_takes_no_mark( assert len(begins) == len(changed) + 1 assert len(_rows_by_file(manager.db_path)["sess-gamma.jsonl"]) == 2 assert _total(manager.db_path) == 4 + + +# ---- overlapping refreshes ------------------------------------------------- + + +def _cold_control(project: Path, tmp_path: Path) -> tuple[dict[str, int], int]: + """Session counts and total of a cache built fresh from the same files.""" + cold = tmp_path / "cold" / project.name + cold.mkdir(parents=True) + for f in project.glob("*.jsonl"): + (cold / f.name).write_bytes(f.read_bytes()) + manager = CacheManager(cold, get_library_version(), db_path=cold.parent / "c.db") + converter.ensure_fresh_cache(cold, manager, silent=True) + data = manager.get_cached_project_data() + assert data is not None + return {sid: s.message_count for sid, s in data.sessions.items()}, ( + data.total_message_count + ) + + +def _overtaken_by( + project: Path, + db_path: Path, + fired: List[int], + errors: List[BaseException], + change: str = "grow", +) -> Any: + """A hook that, once, grows beta (``change="grow"``) or starts a new + session, delta (``"add"``), and runs refresh A to completion on its + own thread (so its own connection), with its own manager.""" + beta = project / "sess-beta.jsonl" + delta = project / "sess-delta.jsonl" + + def refresh_a() -> None: + try: + other = CacheManager(project, get_library_version(), db_path=db_path) + converter.ensure_fresh_cache(project, other, silent=True) + except BaseException as exc: # asserted on B's thread + errors.append(exc) + + def hook(real: Any) -> Any: + def wrapped(self: CacheManager, *args: Any, **kwargs: Any) -> Any: + if not fired: + fired.append(1) + if change == "grow": + _append(beta, _turn("sess-beta", 1, "late beta")) + else: + _append(delta, _turn("sess-delta", 0, "first delta")) + a = threading.Thread(target=refresh_a) + a.start() + a.join(timeout=60) + assert not a.is_alive(), "refresh A is stuck behind refresh B" + return real(self, *args, **kwargs) + + return wrapped + + return hook + + +def _assert_heals_to_the_cold_control( + project: Path, tmp_path: Path, db_path: Path +) -> None: + """After at most one more refresh, B's view is gone: every session + row and the total are what a fresh cache over the files holds.""" + expected = _cold_control(project, tmp_path) + converter.ensure_fresh_cache(project, _manager(project), silent=True) + assert (_sessions(project), _total(db_path)) == expected + assert _manager(project).get_modified_files(sorted(project.glob("*.jsonl"))) == [] + + +def _b_overtaken_at( + hooked: str, + project: Path, + monkeypatch: pytest.MonkeyPatch, + mode: str, + refresh: str, + change: str = "grow", +) -> Path: + """Refresh B, over the built project, of a grown alpha and a new + gamma (two spooled files, so that per file is what the threshold + picks), with refresh A run inside B's ``hooked`` method after the + ``change`` of `_overtaken_by`; returns the cache's path. Asserts that + A ran, cleanly, and the mode B merged in.""" + if refresh == "full": + monkeypatch.setenv("CLAUDE_CODE_LOG_INCREMENTAL_CACHE", "0") + monkeypatch.setenv( + "CLAUDE_CODE_LOG_ATOMIC_MERGE_MB", "0" if mode == "per-file" else "1000" + ) + db_path = project.parent / "cache.db" + monkeypatch.setenv("CLAUDE_CODE_LOG_CACHE_PATH", str(db_path)) + _pass(project.parent) + assert _total(db_path) == 4 + _append(project / "sess-alpha.jsonl", _turn("sess-alpha", 1, "late alpha")) + _append(project / "sess-gamma.jsonl", _turn("sess-gamma", 0, "first gamma")) + + fired: List[int] = [] + errors: List[BaseException] = [] + hook = _overtaken_by(project, db_path, fired, errors, change) + b_thread = threading.get_ident() + b_modes: List[str] = [] + + def recording(name: str, real: Any) -> Any: + def wrapped(self: CacheManager, *args: Any) -> Any: + if threading.get_ident() == b_thread: + b_modes.append(name) + return real(self, *args) + + return wrapped + + with connection_lease(db_path), monkeypatch.context() as m: + m.setattr(CacheManager, hooked, hook(getattr(CacheManager, hooked))) + for name, attr in (("one", "_merge_at_once"), ("per-file", "_merge_per_file")): + m.setattr(CacheManager, attr, recording(name, getattr(CacheManager, attr))) + converter.ensure_fresh_cache(project, _manager(project), silent=True) + assert fired, "refresh A never ran inside B: the test is vacuous" + assert errors == [] + assert b_modes == [mode] + return db_path + + +@pytest.mark.parametrize("refresh", ["full", "incremental"]) +@pytest.mark.parametrize("mode", ["one", "per-file"]) +def test_a_refresh_overtaken_on_a_file_it_read_from_the_cache_heals( + project: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mode: str, + refresh: str, +) -> None: + """Refresh B parses alpha and gamma and takes beta from the cache. + Before B commits, beta grows and refresh A commits it, with the + derived rows. B's derived rows come from the old beta (an + incremental B's total, from the total A replaced): written over A's + record, which matches the file, nothing would ever recompute them.""" + db_path = _b_overtaken_at("_merge_refresh", project, monkeypatch, mode, refresh) + assert _total(db_path) == 0 # B left the project marked unfinished + + _assert_heals_to_the_cold_control(project, tmp_path, db_path) + + +@pytest.mark.parametrize("refresh", ["full", "incremental"]) +@pytest.mark.parametrize("mode", ["one", "per-file"]) +def test_a_refresh_overtaken_by_a_new_file_heals( + project: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mode: str, + refresh: str, +) -> None: + """Before B commits, a new session starts and refresh A commits it. + B never saw delta: its total leaves delta's entries out, under a + record of delta that reads current. A file with a record now and + none as B began has moved, as one whose record changed has.""" + db_path = _b_overtaken_at( + "_merge_refresh", project, monkeypatch, mode, refresh, change="add" + ) + assert _total(db_path) == 0 # B left the project marked unfinished + + _assert_heals_to_the_cold_control(project, tmp_path, db_path) + + +def test_a_cold_build_overtaken_on_a_file_it_wrote_heals( + project: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The cold build writes each file as it parses it. Before its + derived rows, beta grows and refresh A (over the files the build has + written, so with a spool) commits it: the build's derived rows come + from the old beta, over A's current record.""" + db_path = project.parent / "cache.db" + monkeypatch.setenv("CLAUDE_CODE_LOG_CACHE_PATH", str(db_path)) + + fired: List[int] = [] + errors: List[BaseException] = [] + hook = _overtaken_by(project, db_path, fired, errors) + with connection_lease(db_path), monkeypatch.context() as m: + m.setattr( + CacheManager, + "update_session_cache", + hook(CacheManager.update_session_cache), + ) + converter.ensure_fresh_cache(project, _manager(project), silent=True) + assert fired, "refresh A never ran inside the build: the test is vacuous" + assert errors == [] + assert _total(db_path) == 0 # the build left the project unfinished + + _assert_heals_to_the_cold_control(project, tmp_path, db_path) + + +@pytest.mark.parametrize("refresh", ["full", "incremental"]) +def test_a_refresh_overtaken_before_it_read_marks_and_heals( + project: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + refresh: str, +) -> None: + """The check's false positive, pinned: refresh A commits after B's + scope read the records and before B read anything. B's reads see + A's rows, so its derived rows would be right; the check cannot tell, + and marks the project. That costs the next run a full refresh, and + nothing else. (B finds alpha and gamma current, as A left them, and + spools nothing: its merge is one transaction whatever the threshold.)""" + db_path = _b_overtaken_at("_create_overlay", project, monkeypatch, "one", refresh) + assert _total(db_path) == 0 + + _assert_heals_to_the_cold_control(project, tmp_path, db_path)