diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index ab89b723..e946baad 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -4,7 +4,7 @@ This guide covers development setup, testing, architecture, and release processe ## Prerequisites -- Python 3.10+ +- Python 3.10+, linked against SQLite 3.35+ (the cache's minimum, `MIN_SQLITE_VERSION` in `claude_code_log/migrations/runner.py`) - [uv](https://docs.astral.sh/uv/) package manager ## Getting Started @@ -646,10 +646,14 @@ 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 - refresh that fails part-way leaves `projects.refresh_pending` set, and the - next run refreshes that project in full. 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 + 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 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/README.md b/README.md index 3c287d7f..b240c0c2 100644 --- a/README.md +++ b/README.md @@ -284,6 +284,13 @@ Markdown export provides a lightweight, portable alternative to HTML: ## Installation +Requires Python 3.10 or later, linked against **SQLite 3.35 or later** for the +cache (`python3 -c "import sqlite3; print(sqlite3.sqlite_version)"` prints the +version yours uses). The Pythons `uv` installs bundle a recent SQLite; a +distribution's system Python links the system's own, which can be older. With +an older one, `claude-code-log` stops with an error naming both versions, and +`--no-cache` converts without the cache. + The recommended way is installing as a `uv` tool: ```bash diff --git a/claude_code_log/cache.py b/claude_code_log/cache.py index 11a24f76..f71ab23d 100644 --- a/claude_code_log/cache.py +++ b/claude_code_log/cache.py @@ -25,6 +25,12 @@ cache_busy_timeout, run_migrations, ) +from .refresh_spool import ( + MESSAGE_COLUMN_LIST, + MESSAGE_COLUMN_NAMES, + RefreshSpool, + SpooledFile, +) from .write_queue import write_turn from .models import ( AssistantTranscriptEntry, @@ -173,10 +179,6 @@ class ProjectCache(BaseModel): # Session metadata sessions: dict[str, SessionCacheData] - # An entry write has not yet been followed by the session rows and - # aggregates derived from it (migration 015). - refresh_pending: bool = False - # Working directories associated with this project working_directories: list[str] = [] @@ -932,7 +934,9 @@ def _drop_leases(db_path: Path) -> None: """ leases = _thread_leases() for key in [k for k in leases if k[0] == str(db_path)]: - leases.pop(key).close() + conn = leases.pop(key) + conn.close() + _forget_leftover_spool(conn) @contextmanager @@ -973,6 +977,7 @@ def connection_lease( # `_drop_leases` may already have removed and closed it. if leases.pop(key, None) is not None: conn.close() + _forget_leftover_spool(conn) # How long a WAL checkpoint waits for readers to move off old snapshots. A @@ -1118,6 +1123,87 @@ def trim_oversized_wal( return size +# One file's rows, as `_serialize_entry` produces them (`project_id` and +# `file_id` included), into the shared `messages` table. +_INSERT_MESSAGE = ( + f"INSERT INTO messages (project_id, file_id, {MESSAGE_COLUMN_LIST})" + " VALUES (:project_id, :file_id, " + + ", ".join(f":{name}" for name in MESSAGE_COLUMN_NAMES) + + ")" +) + +# A file's record, by (project, name): kept on conflict so its id, and the +# messages rows pointing at it, survive a rewrite (no cascade delete). +_UPSERT_CACHED_FILE = """ + INSERT INTO cached_files + (project_id, file_name, file_path, source_mtime, source_size, + cached_mtime, message_count, subagents_fingerprint, content_version) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(project_id, file_name) DO UPDATE SET + file_path = excluded.file_path, + source_mtime = excluded.source_mtime, + source_size = excluded.source_size, + cached_mtime = excluded.cached_mtime, + message_count = excluded.message_count, + subagents_fingerprint = excluded.subagents_fingerprint, + content_version = excluded.content_version +""" + +# The same columns on a file whose id is known (an append). +_UPDATE_CACHED_FILE = """ + UPDATE cached_files + SET file_path = ?, source_mtime = ?, source_size = ?, cached_mtime = ?, + message_count = ?, subagents_fingerprint = ?, content_version = ? + WHERE id = ? +""" + +# The same rows from an attached spool, with the ids the merge decides. +_MERGE_MESSAGES = ( + f"INSERT INTO messages (project_id, file_id, {MESSAGE_COLUMN_LIST})" + f" SELECT ?, ?, {MESSAGE_COLUMN_LIST} FROM spool.messages" + " WHERE file_id = ? ORDER BY id" +) + + +# 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 +# connection object, not its address, which a later connection may +# reuse. See `CacheManager._detach_spool`. +_leftover_spools: Dict[sqlite3.Connection, Path] = {} + + +def _forget_leftover_spool(conn: sqlite3.Connection) -> None: + """After ``conn`` closed: its attachment went with it, the file goes.""" + path = _leftover_spools.pop(conn, None) + if path is not None: + try: + path.unlink() + except OSError: + pass + + +class _PendingRefresh: + """What an `atomic_refresh` scope accumulates until its merge. + + ``spool`` holds the entry rows. ``deferred`` holds the derived + writes (session rows, sidecar, aggregates) as callables of the + connection, in the order they were requested, run inside the merge + transaction after the rows. ``appends`` keeps, per file spooled as + a tail, the parse that produced it: if the merge finds the file's + 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. + """ + + def __init__(self, spool: RefreshSpool) -> None: + self.spool = spool + self.deferred: List[Callable[[sqlite3.Connection], None]] = [] + self.appends: Dict[ + str, Tuple[Path, List[TranscriptEntry], Optional[str], os.stat_result] + ] = {} + + class CacheManager: """SQLite-based cache manager for Claude Code Log.""" @@ -1162,6 +1248,10 @@ def __init__( # if migrations are ever routed through it. self._shared_conn: Optional[sqlite3.Connection] = None + # 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._project_id: Optional[int] = None if read_only: # Migrations and the project-row upsert both write; a reader @@ -1268,6 +1358,308 @@ def batch(self) -> Generator[None, None, None]: finally: self._shared_conn = None conn.close() + _forget_leftover_spool(conn) + + @contextmanager + def atomic_refresh(self) -> Generator[None, None, None]: + """Commit every write of the scope in one transaction, at its end. + + Inside the scope, ``save_cached_entries`` and + ``extend_cached_entries`` serialise as usual but write to a + private spool file (``refresh_spool.py``) instead of the cache, + taking no shared lock; ``update_session_cache``, + ``save_session_sidecar``, ``merge_session_sidecar`` and + ``update_project_aggregates`` are deferred. On a clean exit the + merge takes one write transaction, copies the spooled rows in + with ``INSERT ... SELECT`` from the attached spool, runs the + deferred writes in the order they were requested, and commits. + An exception leaves the cache exactly as it was: nothing was + written to it, and the spool is removed either way. + + 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. + + Needs a single connection for the scope, so it opens a + ``batch()`` of its own when the caller has none. Nesting joins + the outer scope. A read-only instance cannot write and gets no + spool: its writes behave as outside a scope. + """ + if self._refresh is not None or self._read_only or self._project_id is None: + # Nested: join. Read-only, or no project row: every write of + # this instance returns early, so there is nothing to spool. + yield + return + + with self.batch(), self._get_connection() as conn: + # ATTACH and DETACH refuse to run inside a transaction; the + # scope opens on a connection that is only reading. Python's + # sqlite3 opens one implicitly at any DML outside + # `write_transaction`, so say what happened rather than let + # ATTACH report it. + if conn.in_transaction: + raise sqlite3.ProgrammingError( + "atomic_refresh: the connection is already in a " + "transaction; a write on it bypassed write_transaction" + ) + self._release_leftover_spool(conn) + pending = _PendingRefresh(RefreshSpool(self.db_path)) + try: + conn.execute("ATTACH DATABASE ? AS spool", (str(pending.spool.path),)) + try: + self._refresh = pending + try: + self._create_overlay(conn) + yield + finally: + self._refresh = None + self._drop_overlay(conn) + self._merge_refresh(conn, pending) + finally: + self._detach_spool(conn, pending.spool) + finally: + pending.spool.close() + if conn not in _leftover_spools: + pending.spool.unlink() + + @staticmethod + def _detach_spool(conn: sqlite3.Connection, spool: RefreshSpool) -> None: + """DETACH the spool, or leave it for the next scope on ``conn``. + + A statement still active on the connection (a cursor the scope's + body left unexhausted, kept alive by the traceback of the + exception that ended the scope) makes DETACH fail with "database + spool is locked". That must neither replace the body's exception + nor turn a committed merge into a failure, so the attachment is + recorded and released at the next scope's entry, or at the + connection's close, whichever comes first; the file goes with it. + """ + try: + conn.execute("DETACH DATABASE spool") + except sqlite3.OperationalError: + _leftover_spools[conn] = spool.path + + @staticmethod + def _release_leftover_spool(conn: sqlite3.Connection) -> None: + """Finish a scope's cleanup that `_detach_spool` could not.""" + path = _leftover_spools.pop(conn, None) + if path is None: + return + conn.execute("DROP VIEW IF EXISTS temp.messages") + conn.execute("DROP VIEW IF EXISTS temp.cached_files") + attached = {row[1] for row in conn.execute("PRAGMA database_list")} + if "spool" in attached: + conn.execute("DETACH DATABASE spool") + try: + path.unlink() + except OSError: + pass + + def _create_overlay(self, conn: sqlite3.Connection) -> None: + """Make the scope's reads see the cache as the merge will leave it. + + Two TEMP views named ``cached_files`` and ``messages`` shadow the + tables for every unqualified read on this connection: each is + the table minus the rows the spooled files replace, UNION ALL + the spool's rows. The incremental refresh reads back the rows it + has just written (file states, uuid owners, the closure's + entries); with the rows in the spool, those reads find them here. + A spooled file that is new to the cache has no id yet, so it + reads as ``-``, which nothing else can collide with; + the merge gives it its real id. A spooled tail follows the file's + rows in the cache, so its row ids are offset past the cache's + last, keeping the file's order. Both views select the table's + own columns, in its order, so ``SELECT *`` readers see the same + shape. + + A write through the shadow fails ("cannot modify ... because it + is a view"): inside a scope, the entry rows go to the spool and + nothing else may write these tables, so that is the right noise. + The views are dropped before the merge, which writes the tables. + """ + pid = int(self._project_id or 0) + spooled = "(SELECT file_name FROM spool.files)" + replaced = ( + "(SELECT c.id FROM main.cached_files c JOIN spool.files f" + f" ON f.file_name = c.file_name WHERE c.project_id = {pid}" + " AND f.mode = 'replace')" + ) + # The join giving a spooled file its cache id when it has one. + known = ( + "LEFT JOIN main.cached_files c" + f" ON c.project_id = {pid} AND c.file_name = f.file_name" + ) + + file_columns = [ + row[1] for row in conn.execute("PRAGMA main.table_info(cached_files)") + ] + file_overlay = {"id": "COALESCE(c.id, -f.id)", "project_id": str(pid)} + conn.execute( + f"""CREATE TEMP VIEW cached_files AS + SELECT {", ".join(file_columns)} FROM main.cached_files + WHERE NOT (project_id = {pid} AND file_name IN {spooled}) + UNION ALL + SELECT {", ".join(file_overlay.get(c, f"f.{c}") for c in file_columns)} + FROM spool.files f {known}""" + ) + + message_columns = [ + row[1] for row in conn.execute("PRAGMA main.table_info(messages)") + ] + message_overlay = { + "id": "s.id + (SELECT COALESCE(MAX(id), 0) FROM main.messages)", + "project_id": str(pid), + "file_id": "COALESCE(c.id, -f.id)", + } + conn.execute( + f"""CREATE TEMP VIEW messages AS + SELECT {", ".join(message_columns)} FROM main.messages + WHERE NOT (project_id = {pid} AND file_id IN {replaced}) + UNION ALL + SELECT {", ".join(message_overlay.get(c, f"s.{c}" if c in MESSAGE_COLUMN_NAMES else "NULL") for c in message_columns)} + FROM spool.messages s JOIN spool.files f ON f.id = s.file_id {known}""" + ) + + @staticmethod + def _drop_overlay(conn: sqlite3.Connection) -> None: + conn.execute("DROP VIEW IF EXISTS temp.messages") + conn.execute("DROP VIEW IF EXISTS temp.cached_files") + + def _merge_refresh( + self, conn: sqlite3.Connection, pending: _PendingRefresh + ) -> None: + """The one transaction an `atomic_refresh` scope ends with. + + Runs with the spool attached and the overlay views dropped, so + the tables it writes are the tables. + """ + spooled = pending.spool.files() + if not spooled and not pending.deferred: + return + with write_transaction(conn): + file_ids: List[int] = [] + for spooled_file in spooled: + file_ids.append(self._merge_spooled_file(conn, spooled_file, pending)) + 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, + ) + + if auto_index_enabled() and not index_maintenance_deferred(): + reindex_files(conn, file_ids, commit=False) + self._update_last_updated(conn) + for write in pending.deferred: + write(conn) + + def _merge_spooled_file( + self, + conn: sqlite3.Connection, + spooled: SpooledFile, + pending: _PendingRefresh, + ) -> int: + """Copy one spooled file's rows into the cache; returns its file id.""" + record = ( + self._project_id, + spooled.file_name, + spooled.file_path, + spooled.source_mtime, + spooled.source_size, + spooled.cached_mtime, + spooled.message_count, + spooled.subagents_fingerprint, + spooled.content_version, + ) + if spooled.mode == "append": + row = conn.execute( + "SELECT id FROM cached_files WHERE project_id = ? AND file_name = ?", + (self._project_id, spooled.file_name), + ).fetchone() + if row is not None: + file_id = row["id"] + count = conn.execute( + "SELECT COUNT(*) AS n FROM messages WHERE file_id = ?", (file_id,) + ).fetchone()["n"] + if count == spooled.expected_existing: + conn.execute(_UPDATE_CACHED_FILE, (*record[2:], file_id)) + conn.execute( + _MERGE_MESSAGES, (self._project_id, file_id, spooled.id) + ) + return file_id + # The rows the tail expected to follow are not there (another + # process rewrote the file's rows since the parse): rewrite + # the file whole from the parse, under the lock, as the + # direct path's refused append falls back to a full save. + jsonl_path, entries, subagents_fp, source_stat = pending.appends[ + spooled.file_name + ] + rows = [self._serialize_entry(entry, 0) for entry in entries] + return self._replace_file_rows( + conn, jsonl_path, rows, source_stat, subagents_fp + ) + + conn.execute(_UPSERT_CACHED_FILE, record) + file_id = self._file_id_under_lock(conn, spooled.file_name) + conn.execute("DELETE FROM messages WHERE file_id = ?", (file_id,)) + conn.execute(_MERGE_MESSAGES, (self._project_id, file_id, spooled.id)) + return file_id + + def _write_or_defer( + self, write: Callable[[sqlite3.Connection], None], *, sidecar: bool = False + ) -> None: + """Run a derived write now, or at the scope's merge when in one. + + A sidecar write deferred to the merge runs in a savepoint: a + failure that is not contention rolls back the sidecar alone, + drops its ``sidecar_state`` marker, and lets the rest of the + refresh commit, which is what ``abandon_session_sidecar`` does + for a direct write that failed (the caller handles that one). + Contention re-raises and fails the whole merge, as it fails + the direct write. + """ + if self._refresh is None: + with self._get_connection() as conn, write_transaction(conn): + write(conn) + return + if sidecar: + self._refresh.deferred.append( + functools.partial(self._guarded_sidecar_write, write) + ) + else: + self._refresh.deferred.append(write) + + def _guarded_sidecar_write( + self, write: Callable[[sqlite3.Connection], None], conn: sqlite3.Connection + ) -> None: + """A deferred sidecar write, in a savepoint of the merge.""" + conn.execute("SAVEPOINT sidecar") + try: + write(conn) + except Exception as exc: + if not conn.in_transaction: + # A full disk, an I/O error or memory exhaustion can + # abort the whole transaction, savepoint included; the + # merge then fails on that cause, not on a missing + # savepoint. + raise + conn.execute("ROLLBACK TO sidecar") + conn.execute("RELEASE sidecar") + if is_busy_database_error(exc): + raise + conn.execute( + "DELETE FROM sidecar_state WHERE project_id = ?", (self._project_id,) + ) + logger.warning("Failed to write session sidecar, dropped it: %s", exc) + else: + conn.execute("RELEASE sidecar") def _init_database(self) -> None: """Create schema if needed using migration runner.""" @@ -1404,24 +1796,6 @@ def _update_last_updated(self, conn: sqlite3.Connection) -> None: (datetime.now().isoformat(), self._project_id), ) - def _mark_entries_written(self, conn: sqlite3.Connection) -> None: - """Stamp an entry write: ``last_updated``, and a refresh now pending. - - Called inside the transaction that writes a file's entry rows, so - the flag commits with them (migration 015). The session rows and - aggregates derived from those entries are written later, in other - transactions; until ``update_project_aggregates`` clears the flag, - ``get_modified_files`` reports every file, and a refresh that - stopped in between is redone instead of looking finished. - """ - if self._project_id is None: - return - - conn.execute( - "UPDATE projects SET last_updated = ?, refresh_pending = 1 WHERE id = ?", - (datetime.now().isoformat(), self._project_id), - ) - def _normalize_timestamp(self, timestamp: Optional[str]) -> Optional[str]: """Normalize timestamp to consistent format for reliable string comparison. @@ -1693,9 +2067,6 @@ def save_cached_entries( if source_stat is None: source_stat = jsonl_path.stat() - source_mtime = source_stat.st_mtime - source_size = source_stat.st_size - cached_mtime = datetime.now().timestamp() if subagents_fp is None: subagents_fp = subagents_fingerprint(jsonl_path) @@ -1707,68 +2078,19 @@ def save_cached_entries( # only known inside the transaction, so it is filled in there. serialized_entries = [self._serialize_entry(entry, 0) for entry in entries] - with self._get_connection() as conn, write_transaction(conn): - # Insert or update file record - # Use ON CONFLICT to preserve file ID and avoid cascade deletes on messages - conn.execute( - """ - INSERT INTO cached_files - (project_id, file_name, file_path, source_mtime, source_size, - cached_mtime, message_count, subagents_fingerprint, - content_version) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(project_id, file_name) DO UPDATE SET - file_path = excluded.file_path, - source_mtime = excluded.source_mtime, - source_size = excluded.source_size, - cached_mtime = excluded.cached_mtime, - message_count = excluded.message_count, - subagents_fingerprint = excluded.subagents_fingerprint, - content_version = excluded.content_version - """, - ( - self._project_id, - jsonl_path.name, - str(jsonl_path), - source_mtime, - source_size, - cached_mtime, - len(entries), - subagents_fp, - content_schema_version(), - ), + if self._refresh is not None: + # Spooled: the merge at the scope's end writes it, in one + # transaction with everything else the refresh produced. + self._refresh.spool.add( + self._file_record(jsonl_path, source_stat, len(entries), subagents_fp), + serialized_entries, + mode="replace", ) + return - # Get the file ID - row = conn.execute( - "SELECT id FROM cached_files WHERE project_id = ? AND file_name = ?", - (self._project_id, jsonl_path.name), - ).fetchone() - file_id = row["id"] - - # Delete existing messages for this file - conn.execute("DELETE FROM messages WHERE file_id = ?", (file_id,)) - - # Insert all entries in a batch - for serialized in serialized_entries: - serialized["file_id"] = file_id - conn.executemany( - """ - INSERT INTO messages ( - project_id, file_id, type, timestamp, session_id, - _uuid, _parent_uuid, _is_sidechain, _user_type, _cwd, _version, - _is_meta, _agent_id, _request_id, - input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens, - _leaf_uuid, _level, _operation, content - ) VALUES ( - :project_id, :file_id, :type, :timestamp, :session_id, - :_uuid, :_parent_uuid, :_is_sidechain, :_user_type, :_cwd, :_version, - :_is_meta, :_agent_id, :_request_id, - :input_tokens, :output_tokens, :cache_creation_tokens, :cache_read_tokens, - :_leaf_uuid, :_level, :_operation, :content - ) - """, - serialized_entries, + with self._get_connection() as conn, write_transaction(conn): + file_id = self._replace_file_rows( + conn, jsonl_path, serialized_entries, source_stat, subagents_fp ) # Keep the full-archive search index in step with the rows we @@ -1790,7 +2112,68 @@ def save_cached_entries( if auto_index_enabled() and not index_maintenance_deferred(): reindex_files(conn, [file_id], commit=False) - self._mark_entries_written(conn) + self._update_last_updated(conn) + + def _file_record( + self, + jsonl_path: Path, + source_stat: os.stat_result, + message_count: int, + subagents_fp: Optional[str], + ) -> Dict[str, Any]: + """A file's ``cached_files`` columns as of a parse, by name.""" + return { + "file_name": jsonl_path.name, + "file_path": str(jsonl_path), + "source_mtime": source_stat.st_mtime, + "source_size": source_stat.st_size, + "cached_mtime": datetime.now().timestamp(), + "message_count": message_count, + "subagents_fingerprint": subagents_fp, + "content_version": content_schema_version(), + } + + def _file_id_under_lock(self, conn: sqlite3.Connection, file_name: str) -> int: + """The id of a file record that was just written on ``conn``.""" + row = conn.execute( + "SELECT id FROM cached_files WHERE project_id = ? AND file_name = ?", + (self._project_id, file_name), + ).fetchone() + return row["id"] + + def _replace_file_rows( + self, + conn: sqlite3.Connection, + jsonl_path: Path, + serialized_entries: List[Dict[str, Any]], + source_stat: os.stat_result, + subagents_fp: Optional[str], + ) -> int: + """Write a file's record and its whole row set, under the caller's + write lock; returns the file id.""" + record = self._file_record( + jsonl_path, source_stat, len(serialized_entries), subagents_fp + ) + conn.execute( + _UPSERT_CACHED_FILE, + ( + self._project_id, + record["file_name"], + record["file_path"], + record["source_mtime"], + record["source_size"], + record["cached_mtime"], + record["message_count"], + record["subagents_fingerprint"], + record["content_version"], + ), + ) + file_id = self._file_id_under_lock(conn, jsonl_path.name) + conn.execute("DELETE FROM messages WHERE file_id = ?", (file_id,)) + for serialized in serialized_entries: + serialized["file_id"] = file_id + conn.executemany(_INSERT_MESSAGE, serialized_entries) + return file_id def extend_cached_entries( self, @@ -1835,6 +2218,30 @@ def extend_cached_entries( # `save_cached_entries`; the file id is filled in under it. appended_rows = [self._serialize_entry(entry, 0) for entry in appended] + if self._refresh is not None: + # Spooled as a tail; the row check runs at the merge, under + # its lock, and a refusal there rewrites the file from the + # entries kept here. A file already spooled in this scope + # cannot take a tail (it has rows of its own in the spool + # the check would not see); it is saved whole instead. + if self._refresh.spool.has(jsonl_path.name): + return False + self._refresh.spool.add( + self._file_record( + jsonl_path, source_stat, len(all_entries), subagents_fp + ), + appended_rows, + mode="append", + expected_existing=expected_existing, + ) + self._refresh.appends[jsonl_path.name] = ( + jsonl_path, + all_entries, + subagents_fp, + source_stat, + ) + return True + with self._get_connection() as conn, write_transaction(conn): # The count and the insert have to be one transaction, and # `write_transaction` opens it with `BEGIN IMMEDIATE`: Python's @@ -1884,40 +2291,24 @@ def _append_under_lock( if count != expected_existing: return False + record = self._file_record( + jsonl_path, source_stat, len(all_entries), subagents_fp + ) conn.execute( - """UPDATE cached_files - SET file_path = ?, source_mtime = ?, source_size = ?, - cached_mtime = ?, message_count = ?, subagents_fingerprint = ?, - content_version = ? - WHERE id = ?""", + _UPDATE_CACHED_FILE, ( - str(jsonl_path), - source_stat.st_mtime, - source_stat.st_size, - datetime.now().timestamp(), - len(all_entries), - subagents_fp, - content_schema_version(), + record["file_path"], + record["source_mtime"], + record["source_size"], + record["cached_mtime"], + record["message_count"], + record["subagents_fingerprint"], + record["content_version"], file_id, ), ) conn.executemany( - """ - INSERT INTO messages ( - project_id, file_id, type, timestamp, session_id, - _uuid, _parent_uuid, _is_sidechain, _user_type, _cwd, _version, - _is_meta, _agent_id, _request_id, - input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens, - _leaf_uuid, _level, _operation, content - ) VALUES ( - :project_id, :file_id, :type, :timestamp, :session_id, - :_uuid, :_parent_uuid, :_is_sidechain, :_user_type, :_cwd, :_version, - :_is_meta, :_agent_id, :_request_id, - :input_tokens, :output_tokens, :cache_creation_tokens, :cache_read_tokens, - :_leaf_uuid, :_level, :_operation, :content - ) - """, - [{**row, "file_id": file_id} for row in appended_rows], + _INSERT_MESSAGE, [{**row, "file_id": file_id} for row in appended_rows] ) # The index still refreshes the whole file: `reindex_files` @@ -1934,7 +2325,7 @@ def _append_under_lock( if auto_index_enabled() and not index_maintenance_deferred(): reindex_files(conn, [file_id], commit=False) - self._mark_entries_written(conn) + self._update_last_updated(conn) return True def update_session_cache(self, session_data: Dict[str, SessionCacheData]) -> None: @@ -1942,7 +2333,7 @@ def update_session_cache(self, session_data: Dict[str, SessionCacheData]) -> Non if self._project_id is None: return - with self._get_connection() as conn, write_transaction(conn): + def write(conn: sqlite3.Connection) -> None: for session_id, data in session_data.items(): conn.execute( """ @@ -1995,6 +2386,8 @@ def update_session_cache(self, session_data: Dict[str, SessionCacheData]) -> Non self._update_last_updated(conn) + self._write_or_defer(write) + def update_project_aggregates( self, total_message_count: int, @@ -2007,15 +2400,14 @@ def update_project_aggregates( ) -> None: """Update project-level aggregate information. - The last write of a complete refresh, after every entry write and - the session rows and sidecar derived from them, so it also clears - the ``refresh_pending`` flag those entry writes set (migration - 015). Call it only once those match every entry row written. + The last write of a refresh, after every entry write and the + session rows and sidecar derived from them; inside an + ``atomic_refresh`` scope it commits with them. """ if self._project_id is None: return - with self._get_connection() as conn, write_transaction(conn): + def write(conn: sqlite3.Connection) -> None: conn.execute( """ UPDATE projects SET @@ -2026,8 +2418,7 @@ def update_project_aggregates( total_cache_read_tokens = ?, earliest_timestamp = ?, latest_timestamp = ?, - last_updated = ?, - refresh_pending = 0 + last_updated = ? WHERE id = ? """, ( @@ -2043,6 +2434,8 @@ def update_project_aggregates( ), ) + self._write_or_defer(write) + def save_session_sidecar(self, sidecar: SessionSidecar) -> None: """Persist the cross-session sidecar (migration 008), wholesale. @@ -2058,7 +2451,7 @@ def save_session_sidecar(self, sidecar: SessionSidecar) -> None: if self._project_id is None or self._read_only: return - with self._get_connection() as conn, write_transaction(conn): + def write(conn: sqlite3.Connection) -> None: pid = self._project_id conn.execute("DELETE FROM session_parents WHERE project_id = ?", (pid,)) conn.execute("DELETE FROM junction_uuids WHERE project_id = ?", (pid,)) @@ -2095,6 +2488,8 @@ def save_session_sidecar(self, sidecar: SessionSidecar) -> None: (pid, datetime.now().isoformat()), ) + self._write_or_defer(write, sidecar=True) + def load_session_sidecar(self) -> Optional[SessionSidecar]: """Load the cross-session sidecar, or None when never populated. @@ -2451,8 +2846,8 @@ def abandon_session_sidecar(self, exc: BaseException) -> None: The failed write rolled back, so the previous sidecar still stands, and a refresh that went on to finish would vouch for it. Contention - re-raises: the refresh fails with ``refresh_pending`` still set, and - the retry or the next run redoes it. Any other failure may recur on + re-raises: the refresh fails, and the retry or the next run redoes + it. Any other failure may recur on every run, so instead of failing the project forever this drops the ``sidecar_state`` row: every reader then declines to the full load, as for a cache that never had a sidecar, until an unfiltered @@ -2462,11 +2857,16 @@ def abandon_session_sidecar(self, exc: BaseException) -> None: raise exc if self._project_id is None or self._read_only: return - with self._get_connection() as conn, write_transaction(conn): + + def write(conn: sqlite3.Connection) -> None: conn.execute( "DELETE FROM sidecar_state WHERE project_id = ?", (self._project_id,) ) + # Deferred inside an `atomic_refresh` scope like the write it + # abandons: a refresh that then fails drops nothing either. + self._write_or_defer(write) + def merge_session_sidecar( self, parent_updates: Dict[str, Optional[Tuple[Optional[str], Optional[str]]]], @@ -2486,7 +2886,8 @@ def merge_session_sidecar( """ if self._project_id is None or self._read_only: return - with self._get_connection() as conn, write_transaction(conn): + + def write(conn: sqlite3.Connection) -> None: pid = self._project_id for sid, value in parent_updates.items(): conn.execute( @@ -2530,6 +2931,8 @@ def merge_session_sidecar( (pid, datetime.now().isoformat()), ) + self._write_or_defer(write, sidecar=True) + def get_working_directories(self) -> List[str]: """Get list of working directories associated with this project. @@ -2550,14 +2953,6 @@ def _get_working_directories(self, conn: sqlite3.Connection) -> List[str]: ).fetchall() return [row["cwd"] for row in rows] - def _refresh_pending(self, conn: sqlite3.Connection) -> bool: - """Whether an entry write awaits its derived rows — see - ``_mark_entries_written``.""" - row = conn.execute( - "SELECT refresh_pending FROM projects WHERE id = ?", (self._project_id,) - ).fetchone() - return bool(row and row["refresh_pending"]) - def get_modified_files(self, jsonl_files: List[Path]) -> List[Path]: """Get list of JSONL files that need to be reprocessed. @@ -2567,13 +2962,6 @@ def get_modified_files(self, jsonl_files: List[Path]) -> List[Path]: subagent-sidecar fingerprints, replacing the per-file query + directory probes that dominated freshness-check time on large archives (issue #12). - - Every file is reported while a refresh is pending (migration 015): - then the question is not whether a file's entry rows are current, - it is whether everything derived from them is, and only redoing the - refresh answers that. Per-file entry freshness (``is_file_cached``, - what loading consults) is unaffected, so that refresh is served - from the cache rather than re-parsed. """ if not jsonl_files: return [] @@ -2581,10 +2969,6 @@ def get_modified_files(self, jsonl_files: List[Path]) -> List[Path]: return list(jsonl_files) with self._get_connection() as conn: - if self._refresh_pending(conn): - # A refresh wrote entries and never finished: the session - # rows may predate them, whatever the files look like. - return list(jsonl_files) rows = conn.execute( "SELECT file_name, source_mtime, source_size," " subagents_fingerprint, content_version" @@ -2654,6 +3038,21 @@ def current_fp(jsonl_file: Path, scope: Optional[frozenset[str]]) -> str: return modified + def cached_message_total(self) -> Optional[int]: + """The project's cached ``total_message_count``; None without a row. + + One query, for a pass's plan: ``get_cached_project_data`` would + load every session row to answer the same question. + """ + if self._project_id is None: + return None + with self._get_connection() as conn: + row = conn.execute( + "SELECT total_message_count FROM projects WHERE id = ?", + (self._project_id,), + ).fetchone() + return int(row["total_message_count"]) if row else None + def get_cached_project_data(self) -> Optional[ProjectCache]: """Get the cached project data if available.""" if self._project_id is None: @@ -2722,7 +3121,6 @@ def get_cached_project_data(self) -> Optional[ProjectCache]: total_cache_creation_tokens=project_row["total_cache_creation_tokens"], total_cache_read_tokens=project_row["total_cache_read_tokens"], sessions=sessions, - refresh_pending=bool(project_row["refresh_pending"]), working_directories=working_directories, earliest_timestamp=project_row["earliest_timestamp"], latest_timestamp=project_row["latest_timestamp"], diff --git a/claude_code_log/cli.py b/claude_code_log/cli.py index 5329aafc..fa9ec029 100644 --- a/claude_code_log/cli.py +++ b/claude_code_log/cli.py @@ -43,7 +43,12 @@ is_corrupt_database_error, trim_oversized_wal, ) -from .migrations.runner import apply_write_pragmas, cache_busy_timeout +from .migrations.runner import ( + SQLiteTooOldError, + apply_write_pragmas, + cache_busy_timeout, + check_sqlite_version, +) from .models import RenderingDepth from .render_pool import resolve_render_jobs from .search import ( @@ -380,6 +385,20 @@ def _clear_provider_output(output_root: Path, output_format: str) -> None: ) +def _require_cache_sqlite(remedy: str = "") -> None: + """Refuse up front, in one line, when SQLite is too old for the cache. + + The migration runner refuses too, but by then the conversion is under + way, and it refuses per CacheManager: a pass prints the same warning + twice per project and runs on without a cache. ``remedy`` names the + way out the calling command has, if any. + """ + try: + check_sqlite_version() + except SQLiteTooOldError as exc: + raise click.ClickException(f"{exc}{remedy}.") from exc + + def _install_stack_dump_signal() -> None: """Make ``kill -USR1 `` print the live Python stack to stderr. @@ -1388,6 +1407,16 @@ def convert( # actually projects. Exempt provider_wholesale from both; keep them verbatim # for every other mode, where they remain correct. will_run_all_projects = all_projects or input_path is None + + # A single transcript file and a provider's one-session export never + # open the cache; everything else does unless told not to. + uses_cache = ( + not no_cache + and not (input_path is not None and input_path.is_file()) + and (provider is None or provider_wholesale) + ) + if uses_cache: + _require_cache_sqlite(", or run with --no-cache") if (expand_paths or filter_path) and tui: click.echo( "Warning: --expand-paths / --filter-path are ignored in --tui mode.", @@ -2282,6 +2311,11 @@ def serve( # Before the refresh and the index build, so neither has to work # through a log a previous interrupted serve left behind (#332). The # refresh would do this itself; --no-convert has nothing else that will. + if not no_convert: + _require_cache_sqlite( + ", or serve the pages already generated with --no-convert" + ) + trim_oversized_wal(get_cache_db_path(projects_path), announce=click.echo) if not no_convert: @@ -2603,6 +2637,7 @@ def watch( f"Warning: --theme is HTML-only; ignoring under --format {output_format}.", err=True, ) + _require_cache_sqlite() projects_path = projects_dir or get_default_projects_dir() root = _resolve_watch_root(input_path, projects_path, all_projects) if root is None: diff --git a/claude_code_log/converter.py b/claude_code_log/converter.py index 61fa1033..dbd26598 100644 --- a/claude_code_log/converter.py +++ b/claude_code_log/converter.py @@ -4435,11 +4435,10 @@ def _incremental_cache_refresh( exceeding a third of the project's files (past which the full load is competitive anyway). Message rows written before a decline are exactly what the full path would write for those files, so a - declined attempt never leaves inconsistent state — the crash-window - posture (per-file commits before fact writes) likewise matches the - full path's, and so does its recovery: those commits set - ``refresh_pending`` (migration 015), which only the closing aggregate - write clears, so a refresh that stops between them is redone whole. + declined attempt never leaves inconsistent state: the caller runs + both inside one ``atomic_refresh`` scope, where the attempt's rows + wait in the spool for the full load to pick up, and nothing commits + until the refresh that finishes does, whole. """ from .cache import CachedFileState from .entry_store import PendingStamp @@ -4789,8 +4788,6 @@ def _survives(uuid: str, sess: str) -> bool: logging.getLogger(__name__).warning( "Failed to merge session sidecar, dropped it: %s", e ) - # Last, because it clears ``refresh_pending``: every write of the - # refresh must have landed before the project may read current. cache_manager.update_project_aggregates( total_message_count=cached_project.total_message_count + d_count @@ -4899,49 +4896,52 @@ def ensure_fresh_cache_detailed( if not needs_update: return CacheRefresh.NONE # Cache is already fresh - # 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 project. Declines (False) - # fall through to the unchanged full load below. - # Not after a refresh that never finished (migration 015): which - # session rows it left behind is unknown, so no closure is bounded. - if ( - modified_files - and cached_project_data is not None - and cached_project_data.total_message_count > 0 - and not cached_project_data.refresh_pending - and _incremental_refresh_enabled() - ): - if _incremental_cache_refresh( - project_dir, - cache_manager, - session_jsonl_files, - modified_files, - silent, - entry_store=entry_store, + # 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(): + # 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 + # project. Declines (False) fall through to the full load + # below, in the same scope: the files the attempt parsed + # are in the spool, where the full load finds them current + # and loads them from, so a decline costs no second parse. + if ( + modified_files + and cached_project_data is not None + and cached_project_data.total_message_count > 0 + and _incremental_refresh_enabled() ): - return CacheRefresh.INCREMENTAL - - # Load and process messages to populate cache - if not silent: - print(f"Updating cache for {project_dir.name}...") - messages, _tree = load_directory_transcripts( - project_dir, - cache_manager, - silent=silent, - # The refresh is a full load of the current tree, so persist - # the cross-session sidecar here too (same batch as the - # per-file writes). This is what lets the streaming path run - # right after a cache update without a second full load. - persist_sidecar=True, - sidecar_required=True, - ) + if _incremental_cache_refresh( + project_dir, + cache_manager, + session_jsonl_files, + modified_files, + silent, + entry_store=entry_store, + ): + return CacheRefresh.INCREMENTAL - # Update cache with fresh data. Limit, shared with every clearing - # write: another process's refresh that sets ``refresh_pending`` - # and fails while this one runs is cleared here; closing that - # needs a write generation, not a boolean. - _update_cache_with_session_data(cache_manager, messages) + # Load and process messages to populate cache + if not silent: + print(f"Updating cache for {project_dir.name}...") + messages, _tree = load_directory_transcripts( + project_dir, + cache_manager, + silent=silent, + # The refresh is a full load of the current tree, so + # persist the cross-session sidecar here too (same + # transaction as the per-file writes). This is what lets + # the streaming path run right after a cache update + # without a second full load. + persist_sidecar=True, + sidecar_required=True, + ) + _update_cache_with_session_data(cache_manager, messages) return CacheRefresh.FULL @@ -5979,11 +5979,6 @@ def render_provider_wholesale( # ``has_provider_token_totals``. if has_provider_token_totals: project_aggregates.update(project_token_totals) - # Clears refresh_pending even after a filtered load: v1 - # re-parses every rollout and recomputes every session row - # each run, and is_transcript_stale catches a filtered - # render, so the flag would only add a re-render to every - # filtered run (it saves entries, which set it, each run). cache.update_project_aggregates(**project_aggregates) session_counts = { sid: sd.message_count for sid, sd in merged_session_data.items() @@ -6462,6 +6457,18 @@ def _plan_project( else: combined_stale = True + # A populated project whose cached total is zero is a cache to + # refresh in full, as `ensure_fresh_cache` reads it (migration 016 + # zeroes a project whose derived rows may predate its entries, to + # keep 015's heal); without this, a run that writes no combined + # output would never reach that refresh, since no file is modified + # and no session stale, and the index would count it at zero. + unrefreshed = ( + cache_manager is not None + and bool(jsonl_files) + and cache_manager.cached_message_total() == 0 + ) + # Determine if we need to do any work. With # `write_combined=False`, the combined-transcript file # isn't produced — its staleness / on-disk presence is @@ -6472,6 +6479,7 @@ def _plan_project( bool(modified_files) or bool(stale_sessions) or combined_stale + or unrefreshed or not output_path.exists() ) else: @@ -6480,6 +6488,7 @@ def _plan_project( needs_work = ( bool(modified_files) or bool(stale_sessions) + or unrefreshed or ( combined_ext == "html" and generate_individual_sessions @@ -6491,9 +6500,12 @@ def _plan_project( if window is not None and cache_manager is not None: # A combined page missing at the destination (a fresh --output) is # written even when empty for the window: the index links it. + # A project migration 016 zeroed heals here too: a user who only + # ever runs windowed passes would otherwise keep it at zero. needs_work = ( bool(modified_files) or (write_combined and not output_path.exists()) + or unrefreshed or cache_manager.has_sessions_in_window(*window) ) outside_window = not needs_work @@ -6630,6 +6642,11 @@ def process_projects_hierarchy( # Inside the lease, so the connection that recovers the log # is the one the pass then keeps. trim_oversized_wal(get_cache_db_path(projects_path)) + # A refresh killed mid-parse leaves its spool behind (nothing + # else does); under the pass lock, those of dead processes go. + from .refresh_spool import sweep_orphan_spools + + sweep_orphan_spools(get_cache_db_path(projects_path)) # No entry write of this pass re-indexes its file for search — # not in the pool (its initializer defers too) and not inline # here: that is tokenising under the database's single write diff --git a/claude_code_log/migrations/016_drop_refresh_pending.sql b/claude_code_log/migrations/016_drop_refresh_pending.sql new file mode 100644 index 00000000..01465b1a --- /dev/null +++ b/claude_code_log/migrations/016_drop_refresh_pending.sql @@ -0,0 +1,22 @@ +-- Retire the refresh_pending flag: a refresh now commits whole or not at all +-- Migration: 016 +-- Description: Drop `projects.refresh_pending` (migration 015), keeping its heal. +-- A cache refresh runs as one transaction (`CacheManager.atomic_refresh`): +-- its entry rows go to a spool while it parses, and the merge at its end +-- commits them together with the session rows, the aggregates and the +-- sidecar. A refresh that stops anywhere leaves the cache as it was, so +-- the flag that marked an unfinished refresh has nothing to mark. +-- +-- A project the flag still marks is one whose derived rows may predate +-- its entry rows: a cache written before 015 and upgraded through it +-- (015 marks every project it finds, since nothing else can tell), or +-- one 015's own runs left pending. Zeroing its message total keeps the +-- heal: a project whose total is zero takes the full refresh on its +-- next run, served from the cached entries since no file is modified, +-- which recomputes every derived row, the same repair at the same price. +-- +-- ALTER TABLE ... DROP COLUMN needs SQLite 3.35 (2021-03). + +UPDATE projects SET total_message_count = 0 WHERE refresh_pending = 1; + +ALTER TABLE projects DROP COLUMN refresh_pending; diff --git a/claude_code_log/migrations/runner.py b/claude_code_log/migrations/runner.py index 7649ffb0..29c450df 100644 --- a/claude_code_log/migrations/runner.py +++ b/claude_code_log/migrations/runner.py @@ -6,7 +6,7 @@ import sqlite3 from datetime import datetime from pathlib import Path -from typing import List, Tuple +from typing import List, Optional, Tuple # The WAL file's size, in bytes, that SQLite truncates it back to whenever @@ -20,6 +20,39 @@ _DEFAULT_BUSY_TIMEOUT_SECONDS = 30.0 +# The oldest SQLite the cache's schema can be built with: migration 016 +# uses `ALTER TABLE ... DROP COLUMN`, which arrived in SQLite 3.35.0. +MIN_SQLITE_VERSION: Tuple[int, int, int] = (3, 35, 0) + + +class SQLiteTooOldError(RuntimeError): + """The SQLite linked into this Python is older than the cache needs. + + Not a `sqlite3.DatabaseError`: nothing is wrong with the cache file, + and `CacheManager` must not mistake this for corruption and delete it. + """ + + +def check_sqlite_version( + version_info: Optional[Tuple[int, ...]] = None, +) -> None: + """Raise `SQLiteTooOldError` when SQLite is older than the cache needs. + + Checked before the migration chain runs, never by its failure: an older + SQLite rejects 016's `DROP COLUMN` with a bare syntax error, after the + `UPDATE` above it has already committed, and since the migration is + never recorded the same failure comes back on every run. + """ + found = sqlite3.sqlite_version_info if version_info is None else version_info + if tuple(found) >= MIN_SQLITE_VERSION: + return + need = ".".join(map(str, MIN_SQLITE_VERSION)) + have = ".".join(map(str, found)) + raise SQLiteTooOldError( + f"the cache needs SQLite {need} or later, and this Python uses " + f"SQLite {have}; use a Python built with a newer SQLite" + ) + def cache_busy_timeout() -> float: """Seconds a cache connection waits on a lock before giving up. @@ -222,7 +255,12 @@ def run_migrations(db_path: Path) -> int: Returns: Number of migrations applied + + Raises: + SQLiteTooOldError: SQLite is older than ``MIN_SQLITE_VERSION``; + raised before the database is opened, so nothing is touched. """ + check_sqlite_version() conn = sqlite3.connect(db_path, timeout=cache_busy_timeout()) # Everything after the connect goes inside the try, so no failure can # leave the handle open. That matters most for the pragmas: they are the diff --git a/claude_code_log/pass_lock.py b/claude_code_log/pass_lock.py index 76835cf4..40a3a505 100644 --- a/claude_code_log/pass_lock.py +++ b/claude_code_log/pass_lock.py @@ -9,7 +9,8 @@ committed tends to re-acquire before a sleeper wakes), so the second pass can lose projects to ``database is locked`` however patient its own workers are. It is also the realistic way two refreshes of one project -overlap and clear each other's ``refresh_pending``. +overlap (each commits whole, so the later one wins and the next pass +repairs what it lost; the lock spares the second pass the work). So a pass takes the **pass lock** first: an exclusive transaction on a sidecar SQLite file next to the cache (``.pass-lock``), held until diff --git a/claude_code_log/refresh_spool.py b/claude_code_log/refresh_spool.py new file mode 100644 index 00000000..64c7b030 --- /dev/null +++ b/claude_code_log/refresh_spool.py @@ -0,0 +1,288 @@ +"""A refresh's entry rows, spooled to a private file until one merge. + +A cache refresh parses each modified transcript and writes its rows, +then derives the session rows, the project aggregates and the +cross-session sidecar from the whole. Written straight into the shared +cache, those are many transactions, and a refresh that stops between +two of them (a lock-out, a kill, a full disk) leaves files recorded as +current over derived rows that predate them. Spooled, the rows go into a +file of their own as they are serialised, with no shared lock held, and +the refresh commits everything in one transaction at its end: it +``ATTACH``es the spool and copies with ``INSERT ... SELECT``, which holds +the shared write lock for the copy alone, with no Python per row. + +The spool is an ordinary SQLite file beside the cache database, named +``.spool--``. It sits beside the cache rather than in the +system temp dir because the temp dir may be RAM, and a spool is of the +order of the modified files' compressed size. Its journal is in memory +and nothing is synced: a spool that outlives its process is garbage, and +the refresh that owned it never merged. ``CacheManager.atomic_refresh`` owns +the lifecycle: open, fill, merge, unlink. +""" + +from __future__ import annotations + +import itertools +import os +import sqlite3 +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional + +SPOOL_INFIX = ".spool-" + +# The columns of ``messages`` a writer fills, in the order the cache +# writes them. ``project_id`` and ``file_id`` are not among them: the +# merge sets both, since a new file's id only exists once its +# ``cached_files`` row is written, inside the merge transaction. +MESSAGE_COLUMNS = ( + ("type", "TEXT NOT NULL"), + ("timestamp", "TEXT"), + ("session_id", "TEXT"), + ("_uuid", "TEXT"), + ("_parent_uuid", "TEXT"), + ("_is_sidechain", "INTEGER DEFAULT 0"), + ("_user_type", "TEXT"), + ("_cwd", "TEXT"), + ("_version", "TEXT"), + ("_is_meta", "INTEGER"), + ("_agent_id", "TEXT"), + ("_request_id", "TEXT"), + ("input_tokens", "INTEGER"), + ("output_tokens", "INTEGER"), + ("cache_creation_tokens", "INTEGER"), + ("cache_read_tokens", "INTEGER"), + ("_leaf_uuid", "TEXT"), + ("_level", "TEXT"), + ("_operation", "TEXT"), + ("content", "BLOB NOT NULL"), +) +MESSAGE_COLUMN_NAMES = tuple(name for name, _ in MESSAGE_COLUMNS) +MESSAGE_COLUMN_LIST = ", ".join(MESSAGE_COLUMN_NAMES) + +_SCHEMA = f""" +CREATE TABLE files ( + id INTEGER PRIMARY KEY, + file_name TEXT NOT NULL UNIQUE, + file_path TEXT NOT NULL, + mode TEXT NOT NULL, + expected_existing INTEGER NOT NULL DEFAULT 0, + source_mtime REAL NOT NULL, + source_size INTEGER NOT NULL, + cached_mtime REAL NOT NULL, + message_count INTEGER NOT NULL, + subagents_fingerprint TEXT, + content_version TEXT +); +CREATE TABLE messages ( + id INTEGER PRIMARY KEY, + file_id INTEGER NOT NULL, + {", ".join(f"{name} {decl}" for name, decl in MESSAGE_COLUMNS)} +); +CREATE INDEX messages_file ON messages(file_id); +-- The lookups the incremental refresh makes while the spool overlays the +-- cache (`CacheManager.atomic_refresh`), so they search here as they do +-- in the cache's own indexes. +CREATE INDEX messages_uuid ON messages(_uuid); +CREATE INDEX messages_parent_uuid ON messages(_parent_uuid); +CREATE INDEX messages_request_id ON messages(_request_id); +CREATE INDEX messages_session ON messages(session_id); +CREATE INDEX messages_type ON messages(type); +""" + +_INSERT_ROW = ( + f"INSERT INTO messages (file_id, {MESSAGE_COLUMN_LIST}) VALUES (:file_id, " + + ", ".join(f":{name}" for name in MESSAGE_COLUMN_NAMES) + + ")" +) + +_counter = itertools.count(1) + + +def spool_path(db_path: Path) -> Path: + """A spool file name of this process's own, beside ``db_path``.""" + return db_path.with_name( + f"{db_path.name}{SPOOL_INFIX}{os.getpid()}-{next(_counter)}" + ) + + +def is_spool_path(path: Path, db_path: Path) -> bool: + """Whether ``path`` is a spool of ``db_path``, this process's or not.""" + return path.name.startswith(f"{db_path.name}{SPOOL_INFIX}") + + +def _spool_owner(path: Path, db_path: Path) -> Optional[int]: + """The pid in a spool's name, or None when the name is not one.""" + rest = path.name[len(db_path.name) + len(SPOOL_INFIX) :] + pid, _, seq = rest.partition("-") + if pid.isdigit() and seq.isdigit(): + return int(pid) + return None + + +def _process_alive(pid: int) -> bool: + """Whether a process with that pid exists (ours counts).""" + if pid == os.getpid(): + return True + if sys.platform == "win32": + import ctypes + + kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined] + handle = kernel32.OpenProcess(0x1000, False, pid) # QUERY_LIMITED_INFORMATION + if not handle: + return False + kernel32.CloseHandle(handle) + return True + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + return True + + +def sweep_orphan_spools(db_path: Path) -> List[Path]: + """Remove the spools beside ``db_path`` whose process is gone. + + A refresh removes its spool when it ends, whichever way; only a + killed process leaves one, and it then costs disk until something + removes it. A spool whose owning process is still alive may be mid + refresh and is left alone. Returns the paths removed. + """ + removed: List[Path] = [] + try: + candidates = list(db_path.parent.glob(f"{db_path.name}{SPOOL_INFIX}*")) + except OSError: + return removed + for path in candidates: + pid = _spool_owner(path, db_path) + if pid is None or _process_alive(pid): + continue + try: + path.unlink() + except OSError: + continue + removed.append(path) + return removed + + +@dataclass(frozen=True) +class SpooledFile: + """What the merge needs to know about one spooled file. + + ``mode`` is ``replace`` (the rows are the file's whole new content) + 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. + """ + + id: int + file_name: str + file_path: str + mode: str + expected_existing: int + source_mtime: float + source_size: int + cached_mtime: float + message_count: int + subagents_fingerprint: Optional[str] + content_version: Optional[str] + + +class RefreshSpool: + """One refresh's spool: open, fill per file, hand to the merge, unlink.""" + + def __init__(self, db_path: Path) -> None: + self.path = spool_path(db_path) + # Journal in memory and nothing synced: the file is scratch + # whose only reader is the merge in this same process, and a + # crash makes it garbage either way. Not journal OFF: that + # leaves ROLLBACK undefined, and `add` relies on it to keep the + # spool whole when a file's rows fail part-way (with the + # journal off, a failure past the page cache corrupts the + # file). Autocommit, so a file's rows are visible to the cache + # connection that attaches it as soon as they are written. + self._conn = sqlite3.connect(self.path, isolation_level=None) + try: + self._conn.execute("PRAGMA journal_mode = MEMORY") + self._conn.execute("PRAGMA synchronous = OFF") + self._conn.executescript(_SCHEMA) + except BaseException: + self.close() + raise + + def has(self, file_name: str) -> bool: + """Whether a file of that name is already spooled.""" + row = self._conn.execute( + "SELECT 1 FROM files WHERE file_name = ?", (file_name,) + ).fetchone() + return row is not None + + def add( + self, + record: Dict[str, Any], + rows: Iterable[Dict[str, Any]], + *, + mode: str, + expected_existing: int = 0, + ) -> int: + """Spool ``rows`` as ``record``'s new content (or tail). + + ``record`` carries the ``cached_files`` columns of the parse. A + file spooled twice is replaced: the later parse is the newer + one, as a later direct write would be. + """ + self._conn.execute("BEGIN") + try: + self._conn.execute( + "DELETE FROM messages WHERE file_id IN" + " (SELECT id FROM files WHERE file_name = ?)", + (record["file_name"],), + ) + self._conn.execute( + "DELETE FROM files WHERE file_name = ?", (record["file_name"],) + ) + cursor = self._conn.execute( + """INSERT INTO files + (file_name, file_path, mode, expected_existing, source_mtime, + source_size, cached_mtime, message_count, + subagents_fingerprint, content_version) + VALUES (:file_name, :file_path, :mode, :expected_existing, + :source_mtime, :source_size, :cached_mtime, + :message_count, :subagents_fingerprint, + :content_version)""", + {**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) + ) + self._conn.execute("COMMIT") + except BaseException: + self._conn.execute("ROLLBACK") + raise + return file_id + + def files(self) -> List[SpooledFile]: + """The spooled files, in the order they were first spooled.""" + 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 + FROM files ORDER BY id""" + ).fetchall() + return [SpooledFile(*row) for row in rows] + + def close(self) -> None: + """Close the spool's own connection (the merge attaches by path).""" + self._conn.close() + + def unlink(self) -> None: + """Remove the file; it must be closed and detached everywhere first.""" + try: + self.path.unlink() + except FileNotFoundError: + pass diff --git a/dev-docs/application_model.md b/dev-docs/application_model.md index 7120c651..7dbda4ca 100644 --- a/dev-docs/application_model.md +++ b/dev-docs/application_model.md @@ -228,8 +228,9 @@ The queue is per pass, so two *passes* over one cache — a second CLI run, or a `serve --watch` tick while a CLI run refreshes the same archive — are, to each other, one tight-loop writer each, and the unfair polling above applies between them (it is also the realistic -way two refreshes of one project overlap and clear each other's -`refresh_pending`). So `process_projects_hierarchy` first takes the +way two refreshes of one project overlap; each commits whole, § 2.3, so +the later one wins and the next pass repairs what it lost). So +`process_projects_hierarchy` first takes the **pass lock** ([`pass_lock.py`](../claude_code_log/pass_lock.py)): an exclusive transaction on a sidecar SQLite file beside the cache (`.pass-lock`), held around the WAL trim, the conversions, the @@ -352,36 +353,51 @@ pre-filter (`load_cached_entries_filtered`) and the renderer's cut while `now`, `2 hours ago` or any explicit time, midnight included, is an instant. -**A refresh that stops part-way is redone, not trusted.** A refresh -commits each file's entry rows in a transaction of its own, then writes -the session rows and project aggregates derived from them. Staleness -reads both halves: a file is modified when its `cached_files` row is -older than the file, and a rendered session is stale when its `sessions` -row disagrees with the `html_cache` row of its last render. So a refresh -that stopped in between (a write lost to `database is locked`, issue -#332, or a kill) 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. Since migration 015, every entry -write sets `projects.refresh_pending` in its own transaction and -`update_project_aggregates`, the last write of a complete refresh, -clears it. Every refresh is a complete one, date-filtered runs included -(above). A refresh, on either path, writes the session -rows and the cross-session sidecar (§ 2.12) before it, and does not -pass over a sidecar write it loses: the rolled-back write leaves the old -sidecar in place, which the closing write would otherwise vouch for. -Contention fails the refresh, to be redone; any other failure, which -could recur on every run, drops the sidecar instead -(`abandon_session_sidecar`), so readers fall back to the full load -until one writes it afresh. While the flag is set, -`get_modified_files()` reports every file, which every freshness gate reads (the hierarchy plan, the session-scoped -and streaming paths, the TUI), and `ensure_fresh_cache` takes the full -refresh rather than the incremental one (§ 2.14), whose closure could not -cover rows a previous run left behind. That full refresh loads entries -through the per-file check (`is_file_cached`), which the flag does not -touch, so it re-parses only what really changed. The flag is -project-wide and boolean: two processes refreshing the same project at -once can still clear each other's, as they could already overwrite each -other's session rows. +**A refresh commits whole or not at all.** A refresh writes each +transcript file's entry rows, then the session rows, project aggregates +and cross-session sidecar (§ 2.12) derived from them. Staleness reads +both halves: a file is modified when its `cached_files` row is older +than the file, and a rendered session is stale when its `sessions` row +disagrees with the `html_cache` row of its last render. When each of +those writes was a transaction of its own, a refresh that stopped in +between (a write lost to `database is locked`, issue #332, or a kill) +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 +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 +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 +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 +the incremental refresh (§ 2.14) reads back the rows it has just +written; a write through them fails, which is the point. A sidecar write +that fails at the merge for a reason other than contention rolls back +to a savepoint and drops the `sidecar_state` marker (as +`abandon_session_sidecar` does for a direct write), so readers fall back +to the full load until a refresh writes it afresh; contention fails the +refresh, to be redone. Migration 016 dropped the flag and kept 015's +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. **A run renders what its refresh committed.** Only the refresh writes entry rows; the render reads them and writes none. A session file can @@ -662,14 +678,27 @@ Current migrations: session-bearing row and *prefer* that to seeking the uuids a query asked for, so three queries had to move that predicate out of SQL and into Python to keep it from making them slower. -- `015_project_refresh_pending.sql` — adds `refresh_pending` to +- `015_project_refresh_pending.sql` — added `refresh_pending` to projects: an entry write whose derived session rows and aggregates - have not been written yet (§ 2.3). Existing rows start set, because a + had not been written yet. Existing rows start set, because a cache written before it may already hold that damage and nothing else can tell: one full refresh per project, from the cached entries, on - the first run after upgrading. For a Codex project that run also + the first run after upgrading. Retired by 016 (below), which keeps + that heal. For a Codex project that run also re-renders every session page, because `render_provider_wholesale` picks the pages to re-render from `get_modified_files`. +- `016_drop_refresh_pending.sql` — drops `refresh_pending`: a refresh + commits whole or not at all (§ 2.3), so nothing marks an unfinished + one. A project 015 still marked gets `total_message_count` zeroed + first, which the next run reads as a cache to refresh in full, from + the cached entries: 015's heal at 015's price. `DROP COLUMN` needs + SQLite 3.35 (2021), which makes 3.35 the cache's minimum + (`MIN_SQLITE_VERSION`): `run_migrations` checks it before opening + the file, since on an older SQLite the `DROP` failed as a bare syntax + error after the `UPDATE` had committed, unrecorded, so again on + every run. The CLI checks it once up front (`convert` unless + `--no-cache` or a single file, `serve` unless `--no-convert`, + `watch`) and exits 1 with one line naming both versions. Recreating-tables migrations toggle `PRAGMA foreign_keys = OFF/ON` around the rebuild to avoid losing rows to cascade-deletes during the diff --git a/test/test_cache.py b/test/test_cache.py index f31b761e..dd0f93aa 100644 --- a/test/test_cache.py +++ b/test/test_cache.py @@ -548,10 +548,6 @@ def test_get_modified_files(self, cache_manager, temp_project_dir, sample_entrie # Cache only one file cache_manager.save_cached_entries(file1, sample_entries) - # Finish the refresh as ensure_fresh_cache would: a bare entry write - # leaves one pending (migration 015), and then every file reads as - # modified, whatever its own row says. - cache_manager.update_project_aggregates(0, 0, 0, 0, 0, "", "") # Check modified files all_files = [file1, file2] diff --git a/test/test_cache_atomic_refresh.py b/test/test_cache_atomic_refresh.py new file mode 100644 index 00000000..a6e4d763 --- /dev/null +++ b/test/test_cache_atomic_refresh.py @@ -0,0 +1,1002 @@ +"""A refresh commits as one transaction, or not at all. + +A cache refresh writes each transcript file's entry rows, then the +session rows, the project aggregates and the cross-session sidecar +derived from them. Written one transaction each, a refresh that stops in +between (a lock-out, a kill, a full disk) leaves files recorded as +current over derived rows that predate them. Inside +``CacheManager.atomic_refresh`` the entry rows go to a private spool +file as they are serialised, the derived writes are deferred, and one +merge transaction at the end commits everything: an interruption at any +point leaves the cache exactly as it was. + +These tests drive the scope directly, with the writes a refresh makes, +and read the cache back through plain SQLite. +""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import threading +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +import pytest + +from claude_code_log import converter +from claude_code_log.cache import ( + CacheManager, + SessionSidecar, + connection_lease, + get_library_version, +) +from claude_code_log.converter import ( + compute_project_aggregates, + compute_session_data, + load_transcript, + process_projects_hierarchy, +) +from claude_code_log.models import TranscriptEntry + +TABLES = ( + "projects", + "cached_files", + "messages", + "sessions", + "session_parents", + "junction_uuids", + "dedup_winners", + "sidecar_state", +) + + +def _user( + sid: str, uid: str, parent: str | None, minute: int, text: str +) -> Dict[str, Any]: + return { + "type": "user", + "uuid": uid, + "parentUuid": parent, + "sessionId": sid, + "timestamp": f"2026-01-01T10:{minute:02d}:00Z", + "isSidechain": False, + "userType": "external", + "cwd": "/tmp", + "version": "1.0.0", + "message": {"role": "user", "content": text}, + } + + +def _assistant( + sid: str, uid: str, parent: str, minute: int, text: str +) -> Dict[str, Any]: + return { + "type": "assistant", + "uuid": uid, + "parentUuid": parent, + "sessionId": sid, + "timestamp": f"2026-01-01T10:{minute:02d}:00Z", + "isSidechain": False, + "userType": "external", + "cwd": "/tmp", + "version": "1.0.0", + "requestId": f"req_{uid}", + "message": { + "id": f"msg_{uid}", + "type": "message", + "role": "assistant", + "model": "claude-3-sonnet", + "content": [{"type": "text", "text": text}], + "stop_reason": "end_turn", + "usage": {"input_tokens": 10, "output_tokens": 5}, + }, + } + + +def _turn(sid: str, n: int, text: str) -> List[Dict[str, Any]]: + parent = f"{sid}-a{n - 1}" if n else None + return [ + _user(sid, f"{sid}-u{n}", parent, 2 * n, text), + _assistant(sid, f"{sid}-a{n}", f"{sid}-u{n}", 2 * n + 1, f"reply {n}"), + ] + + +def _append(path: Path, entries: List[Dict[str, Any]]) -> None: + with path.open("a", encoding="utf-8") as f: + for entry in entries: + f.write(json.dumps(entry) + "\n") + st = path.stat() + os.utime(path, (st.st_atime, st.st_mtime + 10)) + + +def _entries(path: Path) -> List[TranscriptEntry]: + return load_transcript(path, None, silent=True) + + +def _dump(db_path: Path) -> Dict[str, List[Tuple[Any, ...]]]: + """Every row of every table a refresh touches, in a plain connection.""" + conn = sqlite3.connect(db_path) + try: + return { + table: sorted(conn.execute(f"SELECT * FROM {table}").fetchall()) + for table in TABLES + } + finally: + conn.close() + + +def _rows_by_file(db_path: Path) -> Dict[str, List[Tuple[Any, ...]]]: + """Each file's message rows in insertion order, without the ids.""" + conn = sqlite3.connect(db_path) + try: + rows = conn.execute( + """SELECT f.file_name, m.type, m.timestamp, m.session_id, m._uuid, + m._parent_uuid, m._request_id, m.content + FROM messages m JOIN cached_files f ON f.id = m.file_id + ORDER BY f.file_name, m.id""" + ).fetchall() + finally: + conn.close() + by_file: Dict[str, List[Tuple[Any, ...]]] = {} + for row in rows: + by_file.setdefault(row[0], []).append(row[1:]) + return by_file + + +def _spools(db_path: Path) -> List[Path]: + return sorted(db_path.parent.glob(f"{db_path.name}.spool-*")) + + +def _derive(manager: CacheManager, files: List[Path]) -> None: + """The derived writes of a refresh, from the files' current entries.""" + messages = [e for f in files for e in _entries(f)] + manager.update_session_cache(compute_session_data(messages, include_cwd=True)) + manager.save_session_sidecar( + SessionSidecar(parents={}, junctions={}, dedup_winners={}) + ) + manager.update_project_aggregates(**compute_project_aggregates(messages)) + + +def _refresh(manager: CacheManager, files: List[Path], changed: List[Path]) -> None: + """The writes of a refresh in which ``changed`` were (re)parsed.""" + for f in changed: + manager.save_cached_entries(f, _entries(f)) + _derive(manager, files) + + +@pytest.fixture +def project(tmp_path: Path) -> Path: + 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 _manager(project: Path) -> CacheManager: + return CacheManager( + project, get_library_version(), db_path=project.parent / "cache.db" + ) + + +def _grow(project: Path) -> Tuple[List[Path], List[Path]]: + """Append to one session and start another; returns (all, changed).""" + alpha = project / "sess-alpha.jsonl" + gamma = project / "sess-gamma.jsonl" + _append(alpha, _turn("sess-alpha", 1, "late alpha")) + _append(gamma, _turn("sess-gamma", 0, "first gamma")) + return sorted(project.glob("*.jsonl")), [alpha, gamma] + + +def test_a_refresh_that_fails_before_its_merge_leaves_the_cache_as_it_was( + project: Path, +) -> None: + manager = _manager(project) + files = sorted(project.glob("*.jsonl")) + _refresh(manager, files, files) + before = _dump(manager.db_path) + assert before["messages"] and before["sessions"] # the build happened + + files, changed = _grow(project) + with pytest.raises(RuntimeError, match="interrupted"): + with manager.atomic_refresh(): + _refresh(manager, files, changed) + # Positive control: the rows went to a spool, not the cache. + assert len(_spools(manager.db_path)) == 1 + assert _dump(manager.db_path) == before + raise RuntimeError("interrupted") + + assert _dump(manager.db_path) == before + assert _spools(manager.db_path) == [] + + +def test_a_refresh_commits_everything_in_one_transaction_at_its_end( + project: Path, +) -> None: + manager = _manager(project) + files = sorted(project.glob("*.jsonl")) + _refresh(manager, files, files) + before = _dump(manager.db_path) + + files, changed = _grow(project) + begins: List[str] = [] + with manager.atomic_refresh(): + # The scope opened the connection it will merge on. + 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) + # Another connection sees nothing of it yet. + assert _dump(manager.db_path) == before + assert begins == [] + after = _dump(manager.db_path) + + assert len(begins) == 1, begins + assert _spools(manager.db_path) == [] + # And everything of it afterwards: the new file, the appended rows, + # the session rows and the aggregates derived from them. + rows = _rows_by_file(manager.db_path) + assert set(rows) == {f.name for f in files} + assert len(rows["sess-alpha.jsonl"]) == 4 + assert len(rows["sess-gamma.jsonl"]) == 2 + sessions = {row[2]: row for row in after["sessions"]} + assert set(sessions) == {"sess-alpha", "sess-beta", "sess-gamma"} + assert ( + after["projects"][0][ + after_columns(manager.db_path).index("total_message_count") + ] + == 8 + ) + + +def after_columns(db_path: Path) -> List[str]: + conn = sqlite3.connect(db_path) + try: + return [c[1] for c in conn.execute("PRAGMA table_info(projects)")] + finally: + conn.close() + + +def test_a_merged_refresh_writes_what_the_direct_writes_would( + project: Path, tmp_path: Path +) -> None: + """Row for row, the merge is the direct path: same records, same + rows in the same order, same derived rows.""" + manager = _manager(project) + files = sorted(project.glob("*.jsonl")) + _refresh(manager, files, files) + files, changed = _grow(project) + with manager.atomic_refresh(): + _refresh(manager, files, changed) + + # The same end state written directly, into a cache of its own. + 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) + + def records(dump: Dict[str, List[Tuple[Any, ...]]]) -> List[Tuple[Any, ...]]: + # (file_name, file_path, source_mtime, message_count, fp, version) + return sorted( + (r[2], r[3], r[4], r[6], r[7], r[9]) for r in dump["cached_files"] + ) + + assert records(merged) == records(reference) + assert [r[2:] for r in merged["sessions"]] == [r[2:] for r in reference["sessions"]] + # The aggregates, minus the timestamps of the two builds. + 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_a_spooled_tail_whose_rows_moved_underneath_is_rewritten_whole( + project: Path, +) -> None: + """The append's row check runs at the merge, under its lock.""" + manager = _manager(project) + files = sorted(project.glob("*.jsonl")) + _refresh(manager, files, files) + alpha = project / "sess-alpha.jsonl" + old_entries = _entries(alpha) + _append(alpha, _turn("sess-alpha", 1, "late alpha")) + all_entries = _entries(alpha) + appended = all_entries[len(old_entries) :] + + with manager.atomic_refresh(): + assert manager.extend_cached_entries(alpha, all_entries, appended) + # Another process rewrites the file's rows before the merge: the + # tail no longer has the rows it expected to follow. + other = _manager(project) + other.save_cached_entries(alpha, old_entries[:1]) + assert len(_rows_by_file(manager.db_path)["sess-alpha.jsonl"]) == 1 + _derive(manager, files) + + rows = _rows_by_file(manager.db_path)["sess-alpha.jsonl"] + assert len(rows) == len(all_entries) + assert [r[3] for r in rows] == [getattr(e, "uuid", None) for e in all_entries] + + +def test_a_tail_for_a_file_already_spooled_is_refused(project: Path) -> None: + """The caller then saves the file whole, as after a direct refusal.""" + manager = _manager(project) + files = sorted(project.glob("*.jsonl")) + _refresh(manager, files, files) + alpha = project / "sess-alpha.jsonl" + old_entries = _entries(alpha) + _append(alpha, _turn("sess-alpha", 1, "late alpha")) + all_entries = _entries(alpha) + with manager.atomic_refresh(): + manager.save_cached_entries(alpha, old_entries) + assert not manager.extend_cached_entries( + alpha, all_entries, all_entries[len(old_entries) :] + ) + manager.save_cached_entries(alpha, all_entries) + _derive(manager, files) + rows = _rows_by_file(manager.db_path)["sess-alpha.jsonl"] + assert [r[3] for r in rows] == [getattr(e, "uuid", None) for e in all_entries] + + +def test_a_sidecar_that_cannot_be_written_is_dropped_and_the_rest_commits( + project: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Such a failure can recur on every run; it must not fail the + refresh forever. The marker goes, so readers decline to the full + load until a sidecar is written whole, as for a direct write.""" + manager = _manager(project) + files = sorted(project.glob("*.jsonl")) + _refresh(manager, files, files) + assert manager.load_session_sidecar() is not None + files, changed = _grow(project) + + # A winner row SQLite refuses (NOT NULL), past the model's validation. + broken = SessionSidecar.model_construct( + parents={}, junctions={}, dedup_winners={"u": None} + ) + with manager.atomic_refresh(): + for f in changed: + manager.save_cached_entries(f, _entries(f)) + messages = [e for f in files for e in _entries(f)] + manager.update_session_cache(compute_session_data(messages, include_cwd=True)) + manager.save_session_sidecar(broken) + manager.update_project_aggregates(**compute_project_aggregates(messages)) + + assert "dropped it" in caplog.text + assert manager.load_session_sidecar() is None + rows = _rows_by_file(manager.db_path) + assert len(rows["sess-gamma.jsonl"]) == 2 + assert manager.get_modified_files(files) == [] + + +def test_a_sidecar_locked_out_at_the_merge_fails_the_whole_refresh( + project: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Contention is transient: the refresh fails and is redone, with + nothing of it in the cache, rather than committing without the + sidecar.""" + manager = _manager(project) + files = sorted(project.glob("*.jsonl")) + _refresh(manager, files, files) + before = _dump(manager.db_path) + files, changed = _grow(project) + + def locked(conn: sqlite3.Connection) -> None: + raise sqlite3.OperationalError("database is locked") + + def locked_sidecar(self: CacheManager, sidecar: SessionSidecar) -> None: + self._write_or_defer(locked, sidecar=True) # pyright: ignore[reportPrivateUsage] + + monkeypatch.setattr(CacheManager, "save_session_sidecar", locked_sidecar) + with pytest.raises(sqlite3.OperationalError, match="locked"): + with manager.atomic_refresh(): + _refresh(manager, files, changed) + assert _dump(manager.db_path) == before + assert _spools(manager.db_path) == [] + + +def _pass(projects_dir: Path) -> None: + process_projects_hierarchy( + projects_dir, output_format="md", write_combined=False, jobs=1 + ) + + +def _locked(*_args: Any, **_kwargs: Any) -> None: + raise sqlite3.OperationalError("database is locked") + + +def _never_busy(_exc: BaseException) -> bool: + return False + + +def _fail_second_file_write(m: pytest.MonkeyPatch) -> None: + """The entry write of the second file parsed, after the first.""" + seen: List[str] = [] + for name in ("save_cached_entries", "extend_cached_entries"): + original: Any = getattr(CacheManager, name) + + def wrapper( + self: CacheManager, + jsonl_path: Path, + *args: Any, + _original: Any = original, + **kwargs: Any, + ) -> Any: + if seen and str(jsonl_path) not in seen: + _locked() + seen.append(str(jsonl_path)) + return _original(self, jsonl_path, *args, **kwargs) + + m.setattr(CacheManager, name, wrapper) + + +def _fail_session_rows_write(m: pytest.MonkeyPatch) -> None: + m.setattr(CacheManager, "update_session_cache", _locked) + + +def _fail_sidecar_write(m: pytest.MonkeyPatch) -> None: + m.setattr(CacheManager, "save_session_sidecar", _locked) + + +@pytest.mark.parametrize( + "inject", + [_fail_second_file_write, _fail_session_rows_write, _fail_sidecar_write], + ids=["second-file-write", "session-rows-write", "sidecar-write"], +) +def test_an_interrupted_full_refresh_leaves_the_cache_as_it_was( + project: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + inject: Any, +) -> None: + """A pass whose full refresh is locked out at any write. + + Whichever write it is, the project is reported failed and the cache + is what it was before the pass: no file recorded as current, no + session row, no aggregate from the refresh that did not finish. + """ + projects_dir = project.parent + db_path = projects_dir / "cache.db" + monkeypatch.setenv("CLAUDE_CODE_LOG_CACHE_PATH", str(db_path)) + monkeypatch.setenv("CLAUDE_CODE_LOG_INCREMENTAL_CACHE", "0") + _pass(projects_dir) + before = _dump(db_path) + assert before["messages"] # the first pass built the cache + + _grow(project) + capsys.readouterr() + with monkeypatch.context() as m: + inject(m) + m.setattr(converter, "is_busy_database_error", _never_busy) + _pass(projects_dir) + assert "projects failed" in capsys.readouterr().out + assert _dump(db_path) == before + assert _spools(db_path) == [] + + # The next pass redoes the whole of it. + _pass(projects_dir) + rows = _rows_by_file(db_path) + assert len(rows["sess-alpha.jsonl"]) == 4 + assert len(rows["sess-gamma.jsonl"]) == 2 + + +def test_a_file_whose_rows_fail_part_way_leaves_the_spool_whole( + tmp_path: Path, +) -> None: + """``add`` rolls back the file it was writing; the spool stays valid. + + With the journal off, SQLite leaves ROLLBACK undefined, and a + failure past the page cache corrupts the file. A small page cache + makes a short run spill, which is what the journal setting has to + survive. + """ + from claude_code_log.refresh_spool import RefreshSpool + + spool = RefreshSpool(tmp_path / "cache.db") + conn = spool._conn # pyright: ignore[reportPrivateUsage] + conn.execute("PRAGMA cache_size = -64") # 64 KiB + record = { + "file_name": "a.jsonl", + "file_path": "/a.jsonl", + "source_mtime": 1.0, + "source_size": 1, + "cached_mtime": 1.0, + "message_count": 1, + "subagents_fingerprint": None, + "content_version": None, + } + + def rows(n: int, fail_at: int | None = None) -> Any: + for i in range(n): + if i == fail_at: + raise RuntimeError("row %d" % i) + yield {name: None for name in ("timestamp", "session_id")} | { + "type": "user", + "_uuid": f"u{i}", + "_parent_uuid": None, + "_is_sidechain": 0, + "_user_type": None, + "_cwd": None, + "_version": None, + "_is_meta": None, + "_agent_id": None, + "_request_id": None, + "input_tokens": None, + "output_tokens": None, + "cache_creation_tokens": None, + "cache_read_tokens": None, + "_leaf_uuid": None, + "_level": None, + "_operation": None, + "content": b"x" * 2048, + } + + spool.add(record, rows(10), mode="replace") + with pytest.raises(RuntimeError, match="row 400"): + spool.add({**record, "file_name": "b.jsonl"}, rows(401, 400), mode="replace") + + assert [f.file_name for f in spool.files()] == ["a.jsonl"] + assert conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0] == 10 + assert conn.execute("PRAGMA integrity_check").fetchone()[0] == "ok" + spool.close() + spool.unlink() + + +def test_reads_inside_a_scope_see_the_spooled_rows(project: Path) -> None: + """The incremental refresh reads back the rows it has just written: + file states, uuid owners, the closure's entries. Inside a scope they + are in the spool, and the reads find them there, a new file's + under a provisional id.""" + manager = _manager(project) + files = sorted(project.glob("*.jsonl")) + _refresh(manager, files, files) + alpha = project / "sess-alpha.jsonl" + gamma = project / "sess-gamma.jsonl" + old_alpha = _entries(alpha) + old_state = manager.get_file_states(["sess-alpha.jsonl"])["sess-alpha.jsonl"] + files, _changed = _grow(project) + all_alpha = _entries(alpha) + + def read_back() -> None: + states = manager.get_file_states(["sess-alpha.jsonl", "sess-gamma.jsonl"]) + assert len(states["sess-alpha.jsonl"].row_fingerprints) == 4 + # The tail follows the file's rows (its ids read past the + # cache's last), which the incremental path's prefix check needs. + assert states["sess-alpha.jsonl"].row_fingerprints[:2] == ( + old_state.row_fingerprints + ) + assert len(states["sess-gamma.jsonl"].row_fingerprints) == 2 + assert states["sess-gamma.jsonl"].sessions == {"sess-gamma"} + owners = manager.get_uuid_owners(["sess-gamma-u0", "sess-alpha-u1"]) + assert owners["sess-gamma-u0"] == {("sess-gamma", "user")} + assert owners["sess-alpha-u1"] == {("sess-alpha", "user")} + assert manager.get_session_file_map()["sess-gamma"] == {"sess-gamma.jsonl"} + assert manager.is_file_cached(alpha) and manager.is_file_cached(gamma) + loaded = manager.load_cached_entries(gamma) + assert loaded is not None and len(loaded) == 2 + loaded = manager.load_cached_entries(alpha) + assert loaded is not None and [getattr(e, "uuid", None) for e in loaded] == [ + getattr(e, "uuid", None) for e in all_alpha + ] + + with manager.atomic_refresh(): + assert manager.extend_cached_entries( + alpha, all_alpha, all_alpha[len(old_alpha) :] + ) + manager.save_cached_entries(gamma, _entries(gamma)) + read_back() + # Unspooled files read as before, through the same views. + assert manager.is_file_cached(project / "sess-beta.jsonl") + _derive(manager, files) + read_back() + # The provisional id was replaced by a real one at the merge. + assert all( + row[0] > 0 + for row in sqlite3.connect(manager.db_path) + .execute("SELECT id FROM cached_files") + .fetchall() + ) + + +def test_the_overlay_searches_the_indexes_of_both_arms(project: Path) -> None: + """The lookups the incremental refresh makes stay index searches on + the cache's side of the overlay, and on the spool's.""" + manager = _manager(project) + files = sorted(project.glob("*.jsonl")) + _refresh(manager, files, files) + files, changed = _grow(project) + with manager.atomic_refresh(): + for f in changed: + manager.save_cached_entries(f, _entries(f)) + conn = manager._shared_conn # pyright: ignore[reportPrivateUsage] + assert conn is not None + + def plan(sql: str) -> str: + return "\n".join( + row[3] for row in conn.execute("EXPLAIN QUERY PLAN " + sql) + ) + + by_uuid = plan( + "SELECT _uuid, session_id, type FROM messages" + " WHERE project_id = 1 AND _uuid IN ('a', 'b')" + ) + assert "idx_messages_project_uuid" in by_uuid, by_uuid + assert "messages_uuid" in by_uuid, by_uuid + by_file = plan( + "SELECT content FROM messages WHERE file_id IN (1, -1) ORDER BY file_id, id" + ) + assert "idx_messages_file" in by_file, by_file + assert "SCAN main.messages" not in by_file and "SCAN m\n" not in by_file, ( + by_file + ) + _derive(manager, files) + + +def _fail_sidecar_merge(m: pytest.MonkeyPatch) -> None: + m.setattr(CacheManager, "merge_session_sidecar", _locked) + + +@pytest.mark.parametrize( + "inject", + [_fail_second_file_write, _fail_session_rows_write, _fail_sidecar_merge], + ids=["second-file-write", "session-rows-write", "sidecar-merge"], +) +def test_an_interrupted_incremental_refresh_leaves_the_cache_as_it_was( + project: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + inject: Any, +) -> None: + """The incremental path, locked out at any of its writes.""" + projects_dir = project.parent + db_path = projects_dir / "cache.db" + monkeypatch.setenv("CLAUDE_CODE_LOG_CACHE_PATH", str(db_path)) + _pass(projects_dir) + before = _dump(db_path) + + _grow(project) + capsys.readouterr() + with monkeypatch.context() as m: + inject(m) + m.setattr(converter, "is_busy_database_error", _never_busy) + _pass(projects_dir) + out = capsys.readouterr().out + # The pass prints the failure's traceback: the path under test is in it. + assert "_incremental_cache_refresh" in out, out + assert "projects failed" in out, out + assert _dump(db_path) == before + assert _spools(db_path) == [] + + _pass(projects_dir) + rows = _rows_by_file(db_path) + assert len(rows["sess-alpha.jsonl"]) == 4 + assert len(rows["sess-gamma.jsonl"]) == 2 + + +def test_two_refreshes_of_one_project_overlapping_leave_it_repairable( + project: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The later commit wins whole; the cache never records a file as + current over derived rows from an earlier parse of it. + + Refresh A parses the session at one turn; while A is between its + entry write and its derived writes, refresh B (another process's, + here another manager's on another thread) parses a newer turn and + commits. With several transactions, A's derived rows then land over + B's entry rows and the file reads current while the session row + predates it: the newer turn is never rendered. In one transaction + A's snapshot wins whole, its file record is older than the file, and + the next pass repairs it. + + B runs on its own thread so that it opens its own connection: on A's + thread it would be handed A's leased one, whose spool is attached, + and fail on attaching its own, a failure the repair pass hid. A is + the refresh alone, not a whole pass: a pass goes on to render, and + its render loads a file that changed since the refresh and writes + those entries outside any refresh, as current, under the old session + rows. That is a hole of the render, not of the refresh, and it opens + for any file written to during a pass, overlap or not. + """ + projects_dir = project.parent + db_path = projects_dir / "cache.db" + monkeypatch.setenv("CLAUDE_CODE_LOG_CACHE_PATH", str(db_path)) + _pass(projects_dir) + alpha = project / "sess-alpha.jsonl" + _append(alpha, _turn("sess-alpha", 1, "turn one")) + turn_one_size = alpha.stat().st_size + + fired: List[int] = [] + b_errors: List[BaseException] = [] + b_rows: List[int] = [] + original = CacheManager.update_session_cache + + def refresh_b() -> None: + try: + other = CacheManager(project, get_library_version(), db_path=db_path) + converter.ensure_fresh_cache(project, other, silent=True) + # What B committed, read before A's merge replaces it. + b_rows.append(len(_rows_by_file(db_path)["sess-alpha.jsonl"])) + except BaseException as exc: # asserted on A's thread, below + b_errors.append(exc) + + def interleave(self: CacheManager, *args: Any, **kwargs: Any) -> None: + if not fired: + fired.append(1) + _append(alpha, _turn("sess-alpha", 2, "turn two")) + b = threading.Thread(target=refresh_b) + b.start() + b.join(timeout=60) + assert not b.is_alive(), "refresh B is stuck behind refresh A" + original(self, *args, **kwargs) + + # Under the lease a pass holds, so A's connection is the one a + # manager on this thread would be handed. + with connection_lease(db_path), monkeypatch.context() as m: + m.setattr(CacheManager, "update_session_cache", interleave) + refreshed = converter.ensure_fresh_cache( + project, _manager(project), silent=True + ) + assert fired # positive control: B ran inside A + assert b_errors == [] + # B committed its own parse, turn two included: 2 + two turns of 2. + assert b_rows == [6] + assert refreshed is not converter.CacheRefresh.NONE + + # A committed after B, whole: its entry rows, its file record, its + # session row, all from the turn-one parse. The record is older than + # the file, so the file reads modified. + with sqlite3.connect(db_path) as conn: + size, count = conn.execute( + "SELECT source_size, message_count FROM cached_files" + " WHERE file_name = 'sess-alpha.jsonl'" + ).fetchone() + (session_count,) = conn.execute( + "SELECT message_count FROM sessions WHERE session_id = 'sess-alpha'" + ).fetchone() + assert (size, count, session_count) == (turn_one_size, 4, 4) + assert len(_rows_by_file(db_path)["sess-alpha.jsonl"]) == 4 + assert alpha in _manager(project).get_modified_files([alpha]) + + # So the next pass repairs it: B's turn is cached and rendered. + _pass(projects_dir) + cached = CacheManager(project, get_library_version(), db_path=db_path) + data = cached.get_cached_project_data() + assert data is not None + assert data.sessions["sess-alpha"].message_count == 6 + assert data.total_message_count == 6 + 2 + assert "turn two" in (project / "session-sess-alpha.md").read_text(encoding="utf-8") + + +def test_migration_016_heals_a_project_015_left_pending_and_drops_the_flag( + project: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Migration 015 marked every project a cache held before it, since + nothing else can tell whether its derived rows predate its entries; + 016 drops the flag, and keeps that heal by zeroing the marked + project's total, which the next run reads as a cache to refresh in + full, from the cached entries.""" + from claude_code_log.migrations.runner import get_current_version, run_migrations + + projects_dir = project.parent + db_path = projects_dir / "cache.db" + monkeypatch.setenv("CLAUDE_CODE_LOG_CACHE_PATH", str(db_path)) + _pass(projects_dir) + + # The cache as 015 leaves it: the flag set on the project, and + # derived rows that may predate its entries (here: a stale count). + with sqlite3.connect(db_path) as conn: + conn.execute( + "ALTER TABLE projects ADD COLUMN refresh_pending INTEGER NOT NULL DEFAULT 0" + ) + conn.execute("UPDATE projects SET refresh_pending = 1") + conn.execute("UPDATE sessions SET message_count = 1") + conn.execute("DELETE FROM _schema_version WHERE version = 16") + assert run_migrations(db_path) == 1 + with sqlite3.connect(db_path) as conn: + assert get_current_version(conn) == 16 + columns = [row[1] for row in conn.execute("PRAGMA table_info(projects)")] + assert "refresh_pending" not in columns + assert conn.execute("SELECT total_message_count FROM projects").fetchone() == ( + 0, + ) + + manager = _manager(project) + files = sorted(project.glob("*.jsonl")) + assert manager.get_modified_files(files) == [] # no file re-parsed + assert converter.ensure_fresh_cache(project, manager, silent=True) + data = manager.get_cached_project_data() + assert data is not None + assert {sid: s.message_count for sid, s in data.sessions.items()} == { + "sess-alpha": 2, + "sess-beta": 2, + } + assert data.total_message_count == 4 + + +@pytest.mark.parametrize("exit_with", ["return", "exception"]) +def test_a_statement_left_active_over_the_overlay_does_not_break_the_scope( + project: Path, exit_with: str +) -> None: + """A cursor the body left unexhausted keeps a statement active, and + DETACH then fails with "database spool is locked". That must not + fail a merge that committed, nor replace the body's own exception + (which, read as a busy error, would be retried as contention), nor + poison the connection for the next scope, as it would under a + pass's leased connection.""" + manager = _manager(project) + files = sorted(project.glob("*.jsonl")) + _refresh(manager, files, files) + files, changed = _grow(project) + + with manager.batch(): + conn = manager._shared_conn # pyright: ignore[reportPrivateUsage] + assert conn is not None + cursor: Optional[sqlite3.Cursor] = None + try: + with manager.atomic_refresh(): + _refresh(manager, files, changed) + cursor = conn.execute("SELECT _uuid FROM messages") + cursor.fetchone() # active, not exhausted + if exit_with == "exception": + raise RuntimeError("the body's own") + except RuntimeError as exc: + assert exit_with == "exception" and str(exc) == "the body's own" + else: + assert exit_with == "return" + assert len(_rows_by_file(manager.db_path)["sess-gamma.jsonl"]) == 2 + del cursor + # The next scope on the same connection works and commits. + _append(project / "sess-beta.jsonl", _turn("sess-beta", 1, "later")) + with manager.atomic_refresh(): + _refresh(manager, files, [project / "sess-beta.jsonl"]) + assert len(_rows_by_file(manager.db_path)["sess-beta.jsonl"]) == 4 + assert _spools(manager.db_path) == [] + + +@pytest.mark.parametrize( + "write_combined", [False, True], ids=["no-combined", "combined"] +) +def test_a_pass_heals_a_project_016_zeroed_whatever_it_writes( + project: Path, monkeypatch: pytest.MonkeyPatch, write_combined: bool +) -> None: + """016's heal is "a zero total takes the full refresh on the next + run". A hierarchy pass only refreshes a project it plans work for, + and without the combined page, work meant modified files, stale + sessions or a theme change: a zeroed project stayed at zero + messages on the index until one of its files changed.""" + from claude_code_log.migrations.runner import run_migrations + + projects_dir = project.parent + db_path = projects_dir / "cache.db" + monkeypatch.setenv("CLAUDE_CODE_LOG_CACHE_PATH", str(db_path)) + + def run() -> None: + process_projects_hierarchy( + projects_dir, output_format="md", write_combined=write_combined, jobs=1 + ) + + run() + # A healthy cache that 015 marked, as every upgrading cache is: no + # file modified, no session stale, only the flag. + with sqlite3.connect(db_path) as conn: + conn.execute( + "ALTER TABLE projects ADD COLUMN refresh_pending INTEGER NOT NULL DEFAULT 0" + ) + conn.execute("UPDATE projects SET refresh_pending = 1") + conn.execute("DELETE FROM _schema_version WHERE version = 16") + assert run_migrations(db_path) == 1 + + run() + data = _manager(project).get_cached_project_data() + assert data is not None + assert data.total_message_count == 4 + assert {sid: s.message_count for sid, s in data.sessions.items()} == { + "sess-alpha": 2, + "sess-beta": 2, + } + index = (projects_dir / "index.md").read_text(encoding="utf-8") + assert "0 messages" not in index, index + + +@pytest.mark.parametrize( + "write_combined", [False, True], ids=["no-combined", "combined"] +) +def test_a_windowed_pass_heals_a_project_016_zeroed( + project: Path, monkeypatch: pytest.MonkeyPatch, write_combined: bool +) -> None: + """A date-filtered --all-projects run converts the projects active + in its window and plans no work for the others. A project 016 + zeroed that is idle in the window must still be refreshed, or a + user who only ever runs windowed passes keeps it at zero messages + on the index after upgrading.""" + from claude_code_log.migrations.runner import run_migrations + + projects_dir = project.parent + idle = project # its sessions are all on day one + active = projects_dir / "-active" + active.mkdir() + _append(active / "sess-act.jsonl", _turn("sess-act", 0, "day one")) + day_two = _turn("sess-act", 1, "day two") + for entry in day_two: + entry["timestamp"] = entry["timestamp"].replace("01-01", "01-02") + _append(active / "sess-act.jsonl", day_two) + db_path = projects_dir / "cache.db" + monkeypatch.setenv("CLAUDE_CODE_LOG_CACHE_PATH", str(db_path)) + + def run(from_date: Optional[str] = None, to_date: Optional[str] = None) -> None: + process_projects_hierarchy( + projects_dir, + output_format="md", + write_combined=write_combined, + jobs=1, + from_date=from_date, + to_date=to_date, + ) + + run() + with sqlite3.connect(db_path) as conn: + conn.execute( + "ALTER TABLE projects ADD COLUMN refresh_pending INTEGER NOT NULL DEFAULT 0" + ) + conn.execute("UPDATE projects SET refresh_pending = 1") + conn.execute("DELETE FROM _schema_version WHERE version = 16") + assert run_migrations(db_path) == 1 + + run(from_date="2026-01-02", to_date="2026-01-02") + run(from_date="2026-01-02", to_date="2026-01-02") + totals: Dict[str, int] = {} + for p in (idle, active): + data = _manager(p).get_cached_project_data() + assert data is not None + totals[p.name] = data.total_message_count + assert totals == {idle.name: 4, active.name: 4}, totals + + +def test_a_leftover_spool_record_does_not_outlive_its_connection( + project: Path, +) -> None: + """A scope that could not DETACH records its spool for the next + scope on that connection. Keyed by the connection's address, the + record was inherited by whatever connection was next allocated + there, and that scope's entry then DETACHed a spool it never had. + """ + manager = _manager(project) + files = sorted(project.glob("*.jsonl")) + _refresh(manager, files, files) + files, changed = _grow(project) + + with manager.batch(): + conn = manager._shared_conn # pyright: ignore[reportPrivateUsage] + assert conn is not None + with manager.atomic_refresh(): + _refresh(manager, files, changed) + cursor = conn.execute("SELECT _uuid FROM messages") + cursor.fetchone() # active when the scope ends + del cursor + # The batch closed its connection: the leftover went with it. + assert _spools(manager.db_path) == [] + + for n in range(20): + _append(project / "sess-beta.jsonl", _turn("sess-beta", n + 1, f"turn {n}")) + with manager.atomic_refresh(): # a fresh connection each time + _refresh(manager, files, [project / "sess-beta.jsonl"]) + assert len(_rows_by_file(manager.db_path)["sess-beta.jsonl"]) == 42 + assert _spools(manager.db_path) == [] + + +def test_a_nested_scope_joins_the_outer_one(project: Path) -> None: + manager = _manager(project) + files = sorted(project.glob("*.jsonl")) + before = _dump(manager.db_path) + with manager.atomic_refresh(): + with manager.atomic_refresh(): + _refresh(manager, files, files) + assert _dump(manager.db_path) == before + assert len(_spools(manager.db_path)) == 1 + assert _rows_by_file(manager.db_path).keys() == {f.name for f in files} + assert _spools(manager.db_path) == [] diff --git a/test/test_cache_content_version.py b/test/test_cache_content_version.py index e86f4871..dec57aa6 100644 --- a/test/test_cache_content_version.py +++ b/test/test_cache_content_version.py @@ -199,10 +199,6 @@ def cached(self, tmp_path: Path): db = tmp_path / "c.db" cm = CacheManager(project, "test-version", db_path=db) cm.save_cached_entries(project / "s1.jsonl", []) - # Finish the refresh as ensure_fresh_cache would: a bare entry write - # leaves one pending (migration 015), and then every file reads as - # modified, whatever its own row says. - cm.update_project_aggregates(0, 0, 0, 0, 0, "", "") return cm, project / "s1.jsonl", db def test_per_file_check_sees_a_foreign_version(self, cached): diff --git a/test/test_cache_refresh_pending.py b/test/test_cache_interrupted_refresh.py similarity index 84% rename from test/test_cache_refresh_pending.py rename to test/test_cache_interrupted_refresh.py index 8ef70762..3633ec93 100644 --- a/test/test_cache_refresh_pending.py +++ b/test/test_cache_interrupted_refresh.py @@ -1,15 +1,16 @@ """A refresh that fails part-way must not leave the project looking current. -A cache refresh writes each transcript file's entry rows in its own -transaction, and only afterwards the per-session rows and project -aggregates derived from them. Staleness reads both halves: a file is -modified when its ``cached_files`` row is older than the file, and a -rendered session is stale when its ``sessions`` row disagrees with the -``html_cache`` row of its last render. A write that fails in between -(``database is locked`` under ``--all-projects``, issue #332, or a kill) -used to leave the files fresh and the session rows as old as the -renders, so every later run said "cached" and never rendered the -missing messages. +A cache refresh writes each transcript file's entry rows, and the +per-session rows and project aggregates derived from them. Staleness +reads both halves: a file is modified when its ``cached_files`` row is +older than the file, and a rendered session is stale when its +``sessions`` row disagrees with the ``html_cache`` row of its last +render. When each of those writes was a transaction of its own, a write +that failed in between (``database is locked`` under ``--all-projects``, +issue #332, or a kill) left the files fresh and the session rows as old +as the renders, so every later run said "cached" and never rendered the +missing messages. A refresh now commits whole or not at all +(``CacheManager.atomic_refresh``), and a failed one is redone. These tests fail one write of a real refresh with the error SQLite raises under lock contention, then run again cleanly and check that the @@ -200,10 +201,9 @@ def test_a_locked_out_project_ends_up_rendered( """Whichever write the lock-out hits, every session ends up rendered. ``next-run``: the pass's in-pass retry is off, so the project fails - and must be repaired by the next run — issue #332's shape, and the one - the ``refresh_pending`` flag exists for. ``in-pass-retry``: the retry - repairs it within the failing pass itself, which it can only do for a - refresh-phase lock-out through that same flag. + and must be repaired by the next run — issue #332's shape. + ``in-pass-retry``: the retry repairs it within the failing pass + itself. """ project = _build_project(tmp_path) projects_dir = project.parent @@ -252,34 +252,6 @@ def stale() -> List[str]: assert stale() == [], "sessions never re-rendered after the failed refresh" -def test_a_failed_refresh_leaves_every_file_reported_modified( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - """The flag that carries the failure from one run to the next.""" - project = _build_project(tmp_path) - monkeypatch.setenv("CLAUDE_CODE_LOG_CACHE_PATH", str(tmp_path / "cache.db")) - _run(project.parent) - files = sorted(project.glob("*.jsonl")) - - manager = CacheManager(project, get_library_version()) - assert manager.get_modified_files(files) == [] - - # An entry write alone — the derived rows not yet recomputed. - from claude_code_log.converter import load_transcript - - _append(files[0], _turn(files[0].stem, 1, "late")) - manager.save_cached_entries(files[0], load_transcript(files[0], silent=True)) - assert sorted(manager.get_modified_files(files)) == files - - # A fresh manager (a later run) sees the same. - later = CacheManager(project, get_library_version()) - assert sorted(later.get_modified_files(files)) == files - - # Recomputing the derived rows is what clears it. - _run(project.parent) - assert CacheManager(project, get_library_version()).get_modified_files(files) == [] - - def _resumed_session(sid: str, parent_sid: str) -> List[Dict[str, Any]]: """A session resumed from ``parent_sid``'s last turn: a cross-session fact.""" first = _user(sid, f"{sid}-u0", f"{parent_sid}-a0", 20, _late_marker(sid)) @@ -459,24 +431,15 @@ def broken(*_args: Any, **_kwargs: Any) -> None: assert repaired == reference -@pytest.mark.parametrize( - "inject", - [_fail_second_file_write, _fail_session_rows_write], - ids=["second-file-write", "session-rows-write"], -) -def test_a_date_filtered_run_finishes_a_pending_refresh_in_full( +def test_a_date_filtered_run_refreshes_the_cache_in_full( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], - inject: Callable[[pytest.MonkeyPatch, List[str]], None], ) -> None: """A date-filtered run refreshes the cache as an unfiltered one would. - Its window applies to what it renders, not to what it refreshes, so - the refresh it runs over a pending project is a complete one: it may - clear ``refresh_pending``, because the session rows and aggregates it - writes count every message, including those outside the window, and - the next unfiltered run renders them. + Its window applies to what it renders, not to what it refreshes: the + session rows and aggregates it writes count every message, including + those outside the window, and the next unfiltered run renders them. """ project = _build_project(tmp_path) projects_dir = project.parent @@ -489,19 +452,6 @@ def test_a_date_filtered_run_finishes_a_pending_refresh_in_full( entry["timestamp"] = entry["timestamp"].replace("01-01", "01-02") _append(project / f"{sid}.jsonl", late) - fired: List[str] = [] - capsys.readouterr() - with monkeypatch.context() as m: - inject(m, fired) - m.setattr(converter, "is_busy_database_error", _never_busy) - _run(projects_dir) - out = capsys.readouterr().out - # Control: the refresh failed and left the flag set. - assert fired.count("LOCKED") == 1 - assert "projects failed" in out, out - pending = CacheManager(project, get_library_version()).get_cached_project_data() - assert pending is not None and pending.refresh_pending - # A window holding only the first day, not the late turn. process_projects_hierarchy( projects_dir, @@ -512,7 +462,7 @@ def test_a_date_filtered_run_finishes_a_pending_refresh_in_full( to_date="2026-01-01", ) after = CacheManager(project, get_library_version()).get_cached_project_data() - assert after is not None and not after.refresh_pending + assert after is not None assert {sid: s.message_count for sid, s in after.sessions.items()} == { sid: 4 for sid in SESSIONS }, "the rows count the window, not the files" @@ -558,5 +508,5 @@ def test_a_date_filtered_run_does_not_start_a_pending_refresh( ) manager = CacheManager(project, get_library_version()) data = manager.get_cached_project_data() - assert data is not None and not data.refresh_pending + assert data is not None assert manager.get_modified_files(files) == [] diff --git a/test/test_cache_size_freshness.py b/test/test_cache_size_freshness.py index e51afdec..bbecc0a9 100644 --- a/test/test_cache_size_freshness.py +++ b/test/test_cache_size_freshness.py @@ -49,10 +49,6 @@ def project(tmp_path: Path) -> Path: def _cache(project: Path) -> CacheManager: cm = CacheManager(project, "test-version") cm.save_cached_entries(project / "s1.jsonl", []) - # Finish the refresh as ensure_fresh_cache would: a bare entry write - # leaves one pending (migration 015), and then every file reads as - # modified, whatever its own row says. - cm.update_project_aggregates(0, 0, 0, 0, 0, "", "") return cm diff --git a/test/test_cache_sqlite_integrity.py b/test/test_cache_sqlite_integrity.py index d3ac8641..7bf43045 100644 --- a/test/test_cache_sqlite_integrity.py +++ b/test/test_cache_sqlite_integrity.py @@ -705,6 +705,12 @@ class TestConnectionCensus: "holds `BEGIN EXCLUSIVE` as the pass lock; its one write (a " "`user_version` probe that surfaces a read-only file) is rolled back", ), + ("refresh_spool.py", "RefreshSpool.__init__"): ( + 1, + "a refresh's private spool file `.spool--`, not the " + "cache database: journal and sync off, read back only by the " + "merge, which attaches it to the cache connection", + ), } @staticmethod diff --git a/test/test_cli_sqlite_requirement.py b/test/test_cli_sqlite_requirement.py new file mode 100644 index 00000000..c4d7ad3d --- /dev/null +++ b/test/test_cli_sqlite_requirement.py @@ -0,0 +1,101 @@ +"""The CLI refuses, once and up front, an SQLite too old for the cache. + +Without the up-front check the migration runner still refused, but per +CacheManager: a pass printed the same warning twice per project and ran on +without a cache. The version is faked; only `sqlite3.sqlite_version_info` is +read by the check, and every command refused here stops before a pool worker +could start. +""" + +import shutil +import sqlite3 +from pathlib import Path + +import pytest +from click.testing import CliRunner + +from claude_code_log.cli import main + +OLD = (3, 31, 1) +SAMPLE = Path(__file__).parent / "test_data" / "representative_messages.jsonl" + + +@pytest.fixture +def projects(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + root = tmp_path / "projects" + project = root / "-tmp-demo" + project.mkdir(parents=True) + shutil.copy(SAMPLE, project / SAMPLE.name) + monkeypatch.setenv("CLAUDE_CODE_LOG_RENDER_JOBS", "1") + monkeypatch.delenv("CLAUDE_CODE_LOG_CACHE_PATH", raising=False) + return root + + +def _refused(result_output: str, exit_code: int) -> None: + assert exit_code == 1, result_output + lines = [line for line in result_output.splitlines() if line.strip()] + assert len(lines) == 1, result_output + assert lines[0].startswith("Error: the cache needs SQLite 3.35.0 or later") + assert "SQLite 3.31.1" in lines[0] + + +def test_convert_refuses_and_names_no_cache( + projects: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(sqlite3, "sqlite_version_info", OLD) + result = CliRunner().invoke( + main, ["--projects-dir", str(projects), "--all-projects"] + ) + _refused(result.output, result.exit_code) + assert "--no-cache" in result.output + assert not (projects / "claude-code-log-cache.db").exists() + assert not (projects / "index.html").exists() + + +def test_convert_with_no_cache_still_runs( + projects: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(sqlite3, "sqlite_version_info", OLD) + result = CliRunner().invoke( + main, ["--projects-dir", str(projects), "--all-projects", "--no-cache"] + ) + assert result.exit_code == 0, result.output + assert "SQLite" not in result.output + assert (projects / "index.html").exists() + + +def test_a_single_file_needs_no_cache( + projects: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(sqlite3, "sqlite_version_info", OLD) + jsonl = projects / "-tmp-demo" / SAMPLE.name + result = CliRunner().invoke(main, [str(jsonl)]) + assert result.exit_code == 0, result.output + assert jsonl.with_suffix(".html").exists() + + +def test_serve_refuses_and_names_no_convert( + projects: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(sqlite3, "sqlite_version_info", OLD) + result = CliRunner().invoke(main, ["serve", "--projects-dir", str(projects)]) + _refused(result.output, result.exit_code) + assert "--no-convert" in result.output + + +def _past_the_gate(*_args: object, **_kwargs: object) -> None: + raise AssertionError("watch went past its SQLite gate") + + +def test_watch_refuses(projects: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # Past the gate, watch converts once and then loops until interrupted: + # under CliRunner a missing gate would hang the test, not fail it. Both + # steps raise instead, so a missing gate is a fast red. + from claude_code_log import cli + from claude_code_log.watch import WatchEngine + + monkeypatch.setattr(cli, "convert_jsonl_to", _past_the_gate) + monkeypatch.setattr(WatchEngine, "run", _past_the_gate) + monkeypatch.setattr(sqlite3, "sqlite_version_info", OLD) + result = CliRunner().invoke(main, ["watch", str(projects / "-tmp-demo")]) + _refused(result.output, result.exit_code) diff --git a/test/test_migrations.py b/test/test_migrations.py index 983a9e3c..20281b7b 100644 --- a/test/test_migrations.py +++ b/test/test_migrations.py @@ -7,10 +7,13 @@ import pytest from claude_code_log.migrations.runner import ( + MIN_SQLITE_VERSION, + SQLiteTooOldError, _compute_checksum, _ensure_schema_version_table, _parse_migration_number, apply_migration, + check_sqlite_version, get_applied_migrations, get_available_migrations, get_current_version, @@ -491,3 +494,71 @@ def tracking_connect(*args: object, **kwargs: object) -> sqlite3.Connection: # recorded close() call: a closed handle refuses to operate. with pytest.raises(sqlite3.ProgrammingError): conn.execute("SELECT 1") + + +class TestSQLiteVersionRequirement: + """The cache refuses an SQLite too old for its schema, by name. + + Migration 016 uses `ALTER TABLE ... DROP COLUMN` (SQLite 3.35). Run on an + older SQLite it failed with a bare syntax error, never recorded itself, + and failed the same way on every later run. The version is faked here: + only `sqlite3.sqlite_version_info` is read by the check. + """ + + OLD = (3, 31, 1) + + def test_names_the_required_and_the_found_version(self): + with pytest.raises(SQLiteTooOldError) as excinfo: + check_sqlite_version(self.OLD) + message = str(excinfo.value) + assert "SQLite 3.35.0 or later" in message + assert "SQLite 3.31.1" in message + + def test_the_minimum_itself_and_newer_pass(self): + check_sqlite_version(MIN_SQLITE_VERSION) + check_sqlite_version((3, 46, 1)) + check_sqlite_version((4, 0, 0)) + + def test_this_interpreters_sqlite_passes(self): + """Without it every other cache test here would be refusing.""" + check_sqlite_version() + + def test_is_not_a_database_error(self): + """`CacheManager` deletes a cache on corruption-shaped DatabaseErrors; + an old SQLite says nothing about the file.""" + assert not issubclass(SQLiteTooOldError, sqlite3.DatabaseError) + + def test_run_migrations_refuses_before_touching_the_file( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + db_path = tmp_path / "cache.db" + monkeypatch.setattr(sqlite3, "sqlite_version_info", self.OLD) + with pytest.raises(SQLiteTooOldError, match=r"3\.35\.0.*3\.31\.1"): + run_migrations(db_path) + assert not db_path.exists() + + def test_cache_manager_refuses_and_keeps_an_existing_cache( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + from claude_code_log.cache import CacheManager + + db_path = tmp_path / "cache.db" + # Directly, not through a CacheManager: its per-process memo would + # skip the runner, and with it the check, for this path. + run_migrations(db_path) + before = db_path.read_bytes() + + monkeypatch.setattr(sqlite3, "sqlite_version_info", self.OLD) + with pytest.raises(SQLiteTooOldError): + CacheManager(tmp_path / "project", "1.0.0", db_path=db_path) + assert db_path.read_bytes() == before + + def test_minimum_covers_every_drop_column_migration(self): + """A migration using DROP COLUMN needs 3.35; the constant must say so.""" + users = [ + path.name + for _, path in get_available_migrations() + if "DROP COLUMN" in path.read_text(encoding="utf-8").upper() + ] + assert users, "no migration uses DROP COLUMN; is 016 still there?" + assert MIN_SQLITE_VERSION >= (3, 35, 0), users diff --git a/test/test_nested_agents.py b/test/test_nested_agents.py index 42256d95..f5c595d0 100644 --- a/test/test_nested_agents.py +++ b/test/test_nested_agents.py @@ -477,10 +477,6 @@ def test_growing_agent_transcript_invalidates_cached_trunk( cm = CacheManager(proj, "0.0.0-test", db_path=tmp_path / "cache.db") load_transcript(trunk, cache_manager=cm, silent=True) - # Finish the refresh as ensure_fresh_cache would: a bare entry write - # leaves one pending (migration 015), and then every file reads as - # modified, whatever its own row says. - cm.update_project_aggregates(0, 0, 0, 0, 0, "", "") assert cm.is_file_cached(trunk) assert cm.get_modified_files([trunk]) == [] @@ -510,10 +506,6 @@ def test_growing_agent_invalidates_only_its_ancestors(self, tmp_path: Path) -> N cm = CacheManager(proj, "0.0.0-test", db_path=tmp_path / "cache.db") load_transcript(trunk, cache_manager=cm, silent=True) - # Finish the refresh as ensure_fresh_cache would: a bare entry write - # leaves one pending (migration 015), and then every file reads as - # modified, whatever its own row says. - cm.update_project_aggregates(0, 0, 0, 0, 0, "", "") assert all(cm.is_file_cached(f) for f in files.values()) def grow(agent: str) -> None: diff --git a/test/test_refresh_spool.py b/test/test_refresh_spool.py new file mode 100644 index 00000000..335d7348 --- /dev/null +++ b/test/test_refresh_spool.py @@ -0,0 +1,72 @@ +"""The refresh spool file: its name, and the sweep of orphaned ones.""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from claude_code_log.refresh_spool import ( + RefreshSpool, + is_spool_path, + spool_path, + sweep_orphan_spools, +) + +# A pid no process can have: past Linux's default pid_max (4194304), and +# past any Windows pid in practice (they are DWORDs but allocated low). +DEAD_PID = 4_200_000 + + +def test_a_spool_is_named_after_the_cache_and_this_process(tmp_path: Path) -> None: + db = tmp_path / "cache.db" + path = spool_path(db) + assert path.parent == tmp_path + assert path.name.startswith(f"cache.db.spool-{os.getpid()}-") + assert is_spool_path(path, db) + assert not is_spool_path(db, db) + # Each call names a new file. + assert spool_path(db) != path + + +def test_the_sweep_removes_a_dead_processs_spool_and_keeps_a_live_ones( + tmp_path: Path, +) -> None: + db = tmp_path / "cache.db" + dead = tmp_path / f"cache.db.spool-{DEAD_PID}-1" + dead.write_bytes(b"left behind") + alive = RefreshSpool(db) # this process's own, mid refresh + other = tmp_path / "cache.db.spool-notapid-1" + other.write_bytes(b"not a spool name") + unrelated = tmp_path / "other.db.spool-4200000-1" + unrelated.write_bytes(b"another cache's") + + removed = sweep_orphan_spools(db) + + assert removed == [dead] + assert not dead.exists() + assert alive.path.exists() and other.exists() and unrelated.exists() + alive.close() + alive.unlink() + + +def test_a_pass_sweeps_orphaned_spools_before_it_starts( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from claude_code_log.converter import process_projects_hierarchy + + projects_dir = tmp_path / "projects" + project = projects_dir / "-proj" + project.mkdir(parents=True) + (project / "s.jsonl").write_text("", encoding="utf-8") + db = tmp_path / "cache.db" + monkeypatch.setenv("CLAUDE_CODE_LOG_CACHE_PATH", str(db)) + dead = tmp_path / f"cache.db.spool-{DEAD_PID}-7" + dead.write_bytes(b"left behind") + + process_projects_hierarchy( + projects_dir, output_format="md", write_combined=False, jobs=1 + ) + + assert not dead.exists() diff --git a/test/test_render_snapshot.py b/test/test_render_snapshot.py index fc927068..6dbc3ffe 100644 --- a/test/test_render_snapshot.py +++ b/test/test_render_snapshot.py @@ -7,8 +7,9 @@ file's new size and mtime, but the session row keeps the counts the refresh computed. The run's pages then show lines its session row does not count, and the cache calls the file current. Only the pending-refresh -flag the entry write leaves set makes the next run refresh it: without -it, the session row would stay behind until the file is written again. +flag the entry write left set (migration 015, dropped by 016) made the +next run refresh it: without it, the session row would have stayed +behind until the file was written again. """ from __future__ import annotations