From e1509c5c5aae57f2443c17e537464979d684796a Mon Sep 17 00:00:00 2001 From: Christian Boos Date: Thu, 5 Nov 2026 02:45:49 +0100 Subject: [PATCH 1/3] Maintain the search index after the pass, not under the write lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Once an index exists, `save_cached_entries` and `extend_cached_entries` re-indexed the file inside their write transaction, so every project worker decompressed and tokenised its messages while holding the cache database's single write lock. That was the longest thing any writer did under the lock (3.4 s of an 11.2 s save on a 122 MB session), with the rest of the pool queued behind it; it is also what pushes a writer in another process past its busy timeout. A hierarchy pass now defers index maintenance: the parent's inline conversions run inside `search.deferred_index_maintenance()`, the pool initializer (`_init_project_worker`) sets the same switch for every worker, and the entry writes skip `reindex_files`. After the pool, the hold-backs and the lock-out retries have all finished, the pass calls `cache.catch_up_search_index`, which re-indexes every file whose `cached_files.cached_mtime` moved and drops vanished files — one short `write_transaction` per file on the pass's leased connection, before the WAL checkpoint. A busy cache at that point is reported, not fatal: the next run or `serve` start catches up, as `ensure_index` always did. The guarantee is unchanged (a run ends with the index current, and `CLAUDE_CODE_LOG_SEARCH_AUTO_INDEX=0` still opts out of all of it); only where the work lands moves. A single file or directory conversion, with no pass around it, still re-indexes inside the write. Measured on 32 large projects (9 GB, 3,201 transcripts all touched, index present, `--jobs 16`): the parallel phase 555 s → 230 s, the longest `save_cached_entries` 44.8 s → 15.2 s (p90 6.8 s → 2.8 s), and then ~305 s of serial catch-up, so the whole run 555 s → 536 s. Shorter holds, not a shorter worst-case pass; on the daily shape the catch-up is seconds. Co-Authored-By: Claude Fable 5.1 --- CONTRIBUTING.md | 4 +- claude_code_log/cache.py | 65 ++++++++++++++-- claude_code_log/converter.py | 73 ++++++++++------- claude_code_log/search.py | 119 +++++++++++++++++++++++++++- dev-docs/application_model.md | 29 +++++++ docs/archive-search.md | 6 ++ test/test_search.py | 142 ++++++++++++++++++++++++++++++++++ 7 files changed, 405 insertions(+), 33 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 54b077a1..605a7530 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -621,7 +621,9 @@ The tool implements a SQLite-based caching system for performance: - **Concurrency**: WAL mode, one writer at a time. Every write goes through `cache.write_transaction`; in the parallel pass, project workers take turns through a fair cross-process queue (`write_queue.py`), so none gives - up on a sibling. Against other processes connections wait 30 s + up on a sibling. Workers write only rows: an existing search index is + caught up once at the end of the pass, by the parent, not inside each + worker's write transaction. Against other processes connections wait 30 s (`CLAUDE_CODE_LOG_CACHE_TIMEOUT` overrides, in seconds), and a project locked out is retried after the pool rather than failed. A pass compacts an oversized WAL before it starts and truncates the WAL when it ends. A diff --git a/claude_code_log/cache.py b/claude_code_log/cache.py index aa10595c..21f1c6e0 100644 --- a/claude_code_log/cache.py +++ b/claude_code_log/cache.py @@ -1030,6 +1030,49 @@ def checkpoint_wal( conn.close() +def catch_up_search_index(db_path: Path) -> int: + """Bring an existing search index up to date after a hierarchy pass. + + The pass's entry writes leave the index alone (see + `search.index_maintenance_deferred`), so this is where the files they + rewrote get re-indexed: once, serially, on this thread's lease when + there is one, after the pool has exited — so the tokenising never + holds the write lock against a sibling. Returns the number of files + refreshed; 0 when no index exists or maintenance is switched off + (`CLAUDE_CODE_LOG_SEARCH_AUTO_INDEX=0`). + + A cache busy with another process's writer is not worth failing the + pass over: the pages are written, and the next pass or `serve` start + catches the index up. That case is reported and returns 0. + """ + from .search import auto_index_enabled, catch_up_index + + if not auto_index_enabled() or not db_path.exists(): + return 0 + leased = _thread_leases().get((str(db_path), False)) + try: + conn = leased or _open_connection( + db_path, + read_only=False, + configure=functools.partial(_configure_connection, read_only=False), + ) + except sqlite3.Error: + return 0 + try: + return catch_up_index(conn) + except sqlite3.OperationalError as exc: + if not is_busy_database_error(exc): + raise + print( + "Warning: the search index was not brought up to date " + f"({exc}); the next run or `serve` start will do it." + ) + return 0 + finally: + if leased is None: + conn.close() + + def trim_oversized_wal( db_path: Path, *, announce: Callable[[str], None] = print ) -> Optional[int]: @@ -1711,9 +1754,16 @@ def save_cached_entries( # search pay essentially nothing — and those who do never meet a # stale index, because the file's old rows go out with its old # messages rather than waiting for the next server start. - from .search import auto_index_enabled, reindex_files + # Inside a hierarchy pass the tokenising would run under the + # write lock in every worker, so the pass defers it to one + # catch-up at its end (`catch_up_search_index`). + from .search import ( + auto_index_enabled, + index_maintenance_deferred, + reindex_files, + ) - if auto_index_enabled(): + if auto_index_enabled() and not index_maintenance_deferred(): reindex_files(conn, [file_id], commit=False) self._mark_entries_written(conn) @@ -1849,10 +1899,15 @@ def _append_under_lock( # The index still refreshes the whole file: `reindex_files` # deletes and re-adds its rows, and an append-only variant # would need its own correctness argument. Only users who have - # built an index pay it, exactly as before. - from .search import auto_index_enabled, reindex_files + # built an index pay it, exactly as before — and a hierarchy + # pass defers it to its end, as `save_cached_entries` does. + from .search import ( + auto_index_enabled, + index_maintenance_deferred, + reindex_files, + ) - if auto_index_enabled(): + if auto_index_enabled() and not index_maintenance_deferred(): reindex_files(conn, [file_id], commit=False) self._mark_entries_written(conn) diff --git a/claude_code_log/converter.py b/claude_code_log/converter.py index f09c7d5f..1b4ad62f 100644 --- a/claude_code_log/converter.py +++ b/claude_code_log/converter.py @@ -59,6 +59,7 @@ from .cache import ( CacheManager, SessionCacheData, + catch_up_search_index, checkpoint_wal, connection_lease, get_all_cached_projects, @@ -6507,6 +6508,16 @@ def _plan_project( ) +def _init_project_worker(turns: Optional[WriteTurns]) -> None: + """The project pool's initializer: the pass-wide write queue, and no + search-index work under the write lock (the parent catches up once + the pool has exited — `catch_up_search_index`).""" + from .search import defer_index_maintenance + + install_write_turns(turns) + defer_index_maintenance() + + def _convert_project_worker( worker_args: Dict[str, Any], ) -> "tuple[str, float, Optional[str], bool]": @@ -6573,6 +6584,8 @@ def process_projects_hierarchy( it, so a caller can tell a complete pass from one that lost projects (the CLI exits non-zero on the latter). """ + from .search import deferred_index_maintenance + lease = ( connection_lease(get_cache_db_path(projects_path)) if use_cache @@ -6583,32 +6596,40 @@ 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)) - result = _process_projects_hierarchy( - projects_path=projects_path, - from_date=from_date, - to_date=to_date, - use_cache=use_cache, - generate_individual_sessions=generate_individual_sessions, - output_format=output_format, - image_export_mode=image_export_mode, - silent=silent, - page_size=page_size, - depth=depth, - compact=compact, - output_dir=output_dir, - expand_paths=expand_paths, - filter_path=filter_path, - write_combined=write_combined, - no_timestamps=no_timestamps, - no_recaps=no_recaps, - theme=theme, - jobs=jobs, - entry_store=entry_store, - failed_projects=failed_projects, - ) + # 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 + # lock. The pass catches the index up once, below. + with deferred_index_maintenance(): + result = _process_projects_hierarchy( + projects_path=projects_path, + from_date=from_date, + to_date=to_date, + use_cache=use_cache, + generate_individual_sessions=generate_individual_sessions, + output_format=output_format, + image_export_mode=image_export_mode, + silent=silent, + page_size=page_size, + depth=depth, + compact=compact, + output_dir=output_dir, + expand_paths=expand_paths, + filter_path=filter_path, + write_combined=write_combined, + no_timestamps=no_timestamps, + no_recaps=no_recaps, + theme=theme, + jobs=jobs, + entry_store=entry_store, + failed_projects=failed_projects, + ) if use_cache: - # The pass's workers have all exited, so this is the moment the - # WAL can be reset — see `checkpoint_wal` (issue #332). + # The pass's workers have all exited: the files they rewrote + # are re-indexed for search now, serially, with no sibling + # waiting on the lock — and then the WAL can be reset, see + # `checkpoint_wal` (issue #332). + catch_up_search_index(get_cache_db_path(projects_path)) checkpoint_wal(get_cache_db_path(projects_path)) return result @@ -7069,7 +7090,7 @@ def _convert_plan_inline( with ProcessPoolExecutor( max_workers=resolved_jobs, mp_context=ctx, - initializer=install_write_turns, + initializer=_init_project_worker, initargs=(WriteTurns(ctx),), ) as pool: future_to_plan = { diff --git a/claude_code_log/search.py b/claude_code_log/search.py index 7abb83b7..4863e865 100644 --- a/claude_code_log/search.py +++ b/claude_code_log/search.py @@ -38,6 +38,7 @@ import re import sqlite3 import zlib +from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, Callable, Iterable, Iterator, Optional, Sequence, cast @@ -46,6 +47,7 @@ # cycle. The name is prefixed to keep it visually separate from the index # internals defined below — it answers a cache-file question, not an FTS one. from .cache import is_corrupt_database_error as _is_corrupt_database_error +from .cache import write_transaction # --------------------------------------------------------------------------- # Field groups @@ -779,6 +781,118 @@ def auto_index_enabled() -> bool: return value.strip().lower() not in ("0", "false", "no", "off") +# Depth of `deferred_index_maintenance` scopes in this process. A pool +# worker's initializer sets it for the worker's whole life. +_deferred_depth = 0 + + +def index_maintenance_deferred() -> bool: + """Whether entry writes should leave the index to a later `catch_up_index`. + + Inside a hierarchy pass the answer is yes, in the parent and in every + project worker: re-indexing a file means decompressing and tokenising + its messages, and done inside the entry write it runs under the cache + database's single write lock — the longest thing any writer did while + holding it (3.4 s of an 11.2 s save on the largest session measured), + and in a pool it lengthened every sibling's wait. The pass catches up + once at its end instead, on the parent's connection, with the pool + gone. Outside a pass (a single-file conversion, the TUI) nothing is + deferred and the entry write maintains the index as before. + """ + return _deferred_depth > 0 + + +def defer_index_maintenance() -> None: + """Defer index maintenance for the rest of this process's life. + + For a pool worker's initializer: the worker only ever runs pass + conversions, and the parent's catch-up covers them all. + """ + global _deferred_depth + _deferred_depth += 1 + + +@contextmanager +def deferred_index_maintenance() -> Iterator[None]: + """Defer index maintenance for the scope (nests; the outermost restores).""" + global _deferred_depth + _deferred_depth += 1 + try: + yield + finally: + _deferred_depth -= 1 + + +def catch_up_index(conn: sqlite3.Connection) -> int: + """Bring an existing index up to date with `cached_files`; return the + number of files touched (re-indexed, added or removed). + + The end-of-pass counterpart of `reindex_files`: every file the pass + rewrote has a new `cached_files.cached_mtime`, so the same comparison + `ensure_index` makes finds them, and vanished files take their rows + with them. Builds nothing when no index exists, and leaves an index + whose table cannot DELETE to the next `ensure_index`, which rebuilds + it (the same rule `reindex_files` follows). + + One transaction per file, opened with the cache's `write_transaction` + so each hold on the write lock is one file's tokenising and an + interrupted catch-up resumes where it stopped. + """ + if not _index_exists(conn) or not _index_delete_supported(conn): + return 0 + fields = indexed_fields(conn) + if not fields: + return 0 + + indexed_files = { + int(row[0]): float(row[1] or 0.0) + for row in conn.execute(f"SELECT file_id, cached_mtime FROM {FILES_TABLE}") + } + current_files = { + int(row[0]): (int(row[1] or 0), float(row[2] or 0.0)) + for row in conn.execute( + "SELECT id, message_count, cached_mtime FROM cached_files" + ) + } + gone = set(indexed_files) - set(current_files) + todo = [ + file_id + for file_id, (_, mtime) in current_files.items() + if file_id not in indexed_files or abs(indexed_files[file_id] - mtime) >= 1e-6 + ] + if not gone and not todo: + return 0 + + placeholders = ",".join("?" * (len(fields) + 1)) + insert_sql = ( + f"INSERT INTO {FTS_TABLE}(rowid, {', '.join(fields)}) VALUES({placeholders})" + ) + now = datetime.now(timezone.utc).isoformat() + if gone: + with write_transaction(conn): + for file_id in gone: + _delete_file_rows(conn, file_id) + conn.execute(f"DELETE FROM {FILES_TABLE} WHERE file_id = ?", (file_id,)) + for file_id in todo: + with write_transaction(conn): + if file_id in indexed_files: + _delete_file_rows(conn, file_id) + _index_file(conn, file_id, fields, insert_sql) + message_count, mtime = current_files[file_id] + conn.execute( + f"INSERT INTO {FILES_TABLE}" + "(file_id, indexed_at, message_count, cached_mtime) " + "VALUES(?, ?, ?, ?) ON CONFLICT(file_id) DO UPDATE SET " + "indexed_at = excluded.indexed_at, " + "message_count = excluded.message_count, " + "cached_mtime = excluded.cached_mtime", + (file_id, now, message_count, mtime), + ) + with write_transaction(conn): + _refresh_indexed_count(conn) + return len(todo) + len(gone) + + def reindex_files( conn: sqlite3.Connection, file_ids: Sequence[int], @@ -793,7 +907,10 @@ def reindex_files( run `serve` pay one `sqlite_master` lookup and nothing else. Cheap enough for that path: ~180 ms for the largest file in an 8 GB - archive, and only files the cache actually rewrote are touched. + archive, and only files the cache actually rewrote are touched. Not + for a hierarchy pass, though: there the entry write holds the + database's single write lock, so the pass defers to `catch_up_index` + (see `index_maintenance_deferred`). Pass `commit=False` to join the caller's transaction rather than committing on its own. diff --git a/dev-docs/application_model.md b/dev-docs/application_model.md index 183a5a47..0e4b9c58 100644 --- a/dev-docs/application_model.md +++ b/dev-docs/application_model.md @@ -186,6 +186,35 @@ Uncontended, the queue plus `BEGIN IMMEDIATE` add ~10 µs to the smallest write (`update_html_cache`, 46 → 56 µs) and nothing measurable to a session save; whole passes are unchanged at `--jobs 1` and within ~4% on a pooled incremental pass, the price of strict turn-taking. +The queue makes waiting safe; it does not shorten the holds, and the +longest thing a worker did while holding the lock was not the row +insert but maintaining the search index: `save_cached_entries` called +`search.reindex_files` inside its write transaction, so once an index +exists every rewritten file was decompressed and tokenised under the +lock (3.4 s of an 11.2 s save on the largest session measured, 122 MB +/ 46k entries), in every worker, with the rest of the pool queued +behind it. So a hierarchy pass **defers index maintenance** +(`search.deferred_index_maintenance` in the parent, +`search.defer_index_maintenance` in the pool initializer +`_init_project_worker`): the entry writes leave the index alone, and +after `_process_projects_hierarchy` returns — pool exited, hold-backs +and retries done — `cache.catch_up_search_index` re-indexes the files +whose `cached_files.cached_mtime` moved, one short transaction per +file on the pass's leased connection, before the WAL checkpoint. What +this buys is shorter holds, not a shorter pass: on 32 large projects +(9 GB, 3,201 transcripts all touched, index present, `--jobs 16`) the +parallel phase went from 555 s to 230 s and the longest +`save_cached_entries` from 44.8 s to 15.2 s (p90 6.8 s → 2.8 s), but +the serial catch-up then took ~305 s, so the whole run was 555 s +against 536 s. The tokenising is the same work wherever it lands; in +the workers it was paid under the lock, and a hold that long is what +pushes a writer in *another* process (the TUI, a second `serve`) past +its 30 s budget. On the daily shape — a handful of files changed — +the catch-up is seconds. Parallelising it (extract in workers, insert +in the parent) is the open lever if that worst case ever matters. A +single file or directory conversion (`convert_jsonl_to` with no pass +around it) still re-indexes inside the write, where nothing competes +for the lock. `CLAUDE_CODE_LOG_SEARCH_AUTO_INDEX=0` switches both off. `serve` takes the same `--jobs` as the main command. A pool-level failure (e.g. a library caller without the `if __name__ == "__main__"` guard `spawn` needs) degrades to inline sequential processing with a warning rather than aborting. diff --git a/docs/archive-search.md b/docs/archive-search.md index 1c4fbd62..ce4e304b 100644 --- a/docs/archive-search.md +++ b/docs/archive-search.md @@ -160,6 +160,12 @@ The index is derived data: deleting the cache rebuilds both. Once an index exists, ordinary `claude-code-log` runs keep it up to date — you don't have to restart the server to search new conversations. +A run over all projects re-indexes the transcripts it rewrote at its end, +once, after the per-project workers have finished: done inside each +worker's cache writes it held the cache's single write lock while +tokenising, and the other workers waited. Converting a single file or +directory re-indexes as it writes, as before. + That costs about 25% extra on a conversion in which *every* transcript changed (3.3 s → 4.1 s over 7,516 messages); a normal run touches a handful of files and the difference isn't noticeable. Nothing is added for people diff --git a/test/test_search.py b/test/test_search.py index f89e35ff..41db665b 100644 --- a/test/test_search.py +++ b/test/test_search.py @@ -952,6 +952,148 @@ def test_conversion_does_not_create_an_index_for_non_users(tmp_path: Path) -> No conn.close() +# --------------------------------------------------------------------------- +# Maintenance deferred to the end of a hierarchy pass +# --------------------------------------------------------------------------- + + +def test_catch_up_index_refreshes_new_changed_and_removed_files( + cache: sqlite3.Connection, +) -> None: + """The end-of-pass catch-up does what `ensure_index` does for an index + that already exists: new files in, rewritten files replaced, vanished + files out — and reports how many it touched.""" + from claude_code_log.search import catch_up_index + + ensure_index(cache) + assert catch_up_index(cache) == 0, "an unchanged cache has nothing to catch up" + + # A rewritten file: new rows, re-stamped cached_mtime (what the cache does). + cache.execute("DELETE FROM messages WHERE file_id = 1") + _add_message(cache, 30, 1, 1, _assistant("now mentions mercurial"), uuid="uuid-30") + cache.execute("UPDATE cached_files SET cached_mtime = 12345.678 WHERE id = 1") + # A new file. + cache.execute( + "INSERT INTO cached_files(id, project_id, file_name, file_path, " + "source_mtime, cached_mtime, message_count) " + "VALUES(3, 1, 'f3.jsonl', '/tmp/f3.jsonl', 0, 1.0, 1)" + ) + _add_message(cache, 20, 1, 3, _assistant("a brand new file"), uuid="uuid-20") + # A vanished file. + cache.execute("DELETE FROM messages WHERE file_id = 2") + cache.execute("DELETE FROM cached_files WHERE id = 2") + cache.commit() + + assert catch_up_index(cache) == 3, "one rewritten, one new, one vanished" + assert count_matches(cache, "mercurial") == 1 + assert count_matches(cache, "brand") == 1 + assert count_matches(cache, "pydantic") == 0, "stale or vanished rows survived" + assert index_status(cache).indexed_messages == 2 + assert catch_up_index(cache) == 0 + + +def test_catch_up_index_without_an_index_builds_nothing( + cache: sqlite3.Connection, +) -> None: + """Catching up is maintenance of an index someone built; it never creates one.""" + from claude_code_log.search import catch_up_index + + assert catch_up_index(cache) == 0 + row = cache.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?", + (FTS_TABLE,), + ).fetchone() + assert row is None + + +def test_deferred_index_maintenance_is_scoped() -> None: + from claude_code_log.search import ( + deferred_index_maintenance, + index_maintenance_deferred, + ) + + assert not index_maintenance_deferred() + with deferred_index_maintenance(): + assert index_maintenance_deferred() + with deferred_index_maintenance(): + assert index_maintenance_deferred() + assert index_maintenance_deferred(), "an inner scope ended the outer one" + assert not index_maintenance_deferred() + + +def _report_deferred() -> bool: + from claude_code_log.search import index_maintenance_deferred + + return index_maintenance_deferred() + + +def test_project_pool_workers_defer_index_maintenance() -> None: + """A project worker never tokenises under the cache's write lock: the + pool's initializer defers the index work to the parent's catch-up.""" + import multiprocessing + from concurrent.futures import ProcessPoolExecutor + + from claude_code_log.converter import _init_project_worker # pyright: ignore[reportPrivateUsage] + from claude_code_log.write_queue import WriteTurns + + ctx = multiprocessing.get_context("spawn") + with ProcessPoolExecutor( + max_workers=1, + mp_context=ctx, + initializer=_init_project_worker, + initargs=(WriteTurns(ctx),), + ) as pool: + assert pool.submit(_report_deferred).result(timeout=60) is True + + +def test_a_hierarchy_pass_maintains_the_index_after_the_pool_not_inside_the_writes( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The guarantee of `test_a_normal_conversion_keeps_an_existing_index_current` + is kept by one catch-up at the end of the pass, not by every entry + write re-indexing its file inside the write transaction (where the + tokenising held the database's single write lock, in every worker). + With the catch-up stubbed out, the pass must leave the index stale.""" + from claude_code_log import converter + from claude_code_log.cache import get_cache_db_path + from claude_code_log.converter import process_projects_hierarchy + + projects = tmp_path / "projects" + project = projects / "-home-u-proj" + project.mkdir(parents=True) + jsonl = project / "session.jsonl" + jsonl.write_text(_entry_line("the first version mentions kangaroo")) + process_projects_hierarchy(projects, silent=True) + + db_path = get_cache_db_path(projects) + conn = sqlite3.connect(db_path) + ensure_index(conn) + assert count_matches(conn, "kangaroo") == 1 + conn.close() + + calls: list[Path] = [] + + def record(path: Path) -> int: + calls.append(path) + return 0 + + monkeypatch.setattr(converter, "catch_up_search_index", record) + + jsonl.write_text(_entry_line("the second version mentions platypus")) + bump_mtime(jsonl) + process_projects_hierarchy(projects, silent=True) + + assert calls == [db_path], "the pass did not run one catch-up on its cache" + conn = sqlite3.connect(db_path) + try: + assert count_matches(conn, "platypus") == 0, ( + "the entry write maintained the index itself" + ) + assert count_matches(conn, "kangaroo") == 1 + finally: + conn.close() + + def _entry_line(text: str) -> str: """One user message as a JSONL line.""" return ( From 80bdcc268af31c744a178b140a20ed3fc115012c Mon Sep 17 00:00:00 2001 From: Christian Boos Date: Thu, 5 Nov 2026 02:55:58 +0100 Subject: [PATCH 2/3] Treat every catch-up failure as a warning, and clear a file's rows before re-indexing it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review findings on the end-of-pass catch-up. A non-BUSY error during the catch-up escaped the whole pass: a `zlib.error` from one blob, or `database or disk is full`, raised out of `process_projects_hierarchy` with `failed_projects` empty and the WAL checkpoint skipped — where the same error inside an entry write had only failed that one project's save. The index is derived data that the next pass heals (its entries stay behind their cache rows until re-extracted), so `catch_up_search_index` now reports any `sqlite3.Error` or `zlib.error` with the same warning as a busy cache and returns 0, and the converter checkpoints the WAL in a `finally`. Pinned for a zlib error, a full disk and a malformed database: the pass returns, loses no project, checkpoints, and the following pass brings the index current. `catch_up_index` skipped `_delete_file_rows` for a file its snapshot did not know, so a file another process indexed between the snapshot and the transaction had its rowids inserted a second time — which a contentless FTS5 table accepts without a word. The delete is now unconditional (a no-op when there are no rows), in `ensure_index` too, which had the same shape. Measured on SQLite 3.50.4 the duplicate is invisible to MATCH and cleared by one DELETE, so the test for it pins the invariant rather than a visible symptom. Co-Authored-By: Claude Fable 5.1 --- claude_code_log/cache.py | 14 +++-- claude_code_log/converter.py | 10 +++- claude_code_log/search.py | 15 +++-- test/test_search.py | 107 +++++++++++++++++++++++++++++++++++ 4 files changed, 132 insertions(+), 14 deletions(-) diff --git a/claude_code_log/cache.py b/claude_code_log/cache.py index 21f1c6e0..0535f263 100644 --- a/claude_code_log/cache.py +++ b/claude_code_log/cache.py @@ -1041,9 +1041,13 @@ def catch_up_search_index(db_path: Path) -> int: refreshed; 0 when no index exists or maintenance is switched off (`CLAUDE_CODE_LOG_SEARCH_AUTO_INDEX=0`). - A cache busy with another process's writer is not worth failing the - pass over: the pages are written, and the next pass or `serve` start - catches the index up. That case is reported and returns 0. + The index is derived data that the next pass heals (its entries stay + behind their cache rows until re-extracted), so no error here is + worth failing the pass over: the pages are written already. A cache + busy with another process's writer, a full disk, a damaged index and + an undecodable blob are all reported with the same warning and + return 0 — unlike an entry write, where the same error fails that + one project's save and nothing else. """ from .search import auto_index_enabled, catch_up_index @@ -1060,9 +1064,7 @@ def catch_up_search_index(db_path: Path) -> int: return 0 try: return catch_up_index(conn) - except sqlite3.OperationalError as exc: - if not is_busy_database_error(exc): - raise + except (sqlite3.Error, zlib.error) as exc: print( "Warning: the search index was not brought up to date " f"({exc}); the next run or `serve` start will do it." diff --git a/claude_code_log/converter.py b/claude_code_log/converter.py index 1b4ad62f..67ff583c 100644 --- a/claude_code_log/converter.py +++ b/claude_code_log/converter.py @@ -6628,9 +6628,13 @@ def process_projects_hierarchy( # The pass's workers have all exited: the files they rewrote # are re-indexed for search now, serially, with no sibling # waiting on the lock — and then the WAL can be reset, see - # `checkpoint_wal` (issue #332). - catch_up_search_index(get_cache_db_path(projects_path)) - checkpoint_wal(get_cache_db_path(projects_path)) + # `checkpoint_wal` (issue #332). The catch-up reports its own + # failures; the `finally` is for whatever it did not foresee, + # so the checkpoint is never the casualty of the index. + try: + catch_up_search_index(get_cache_db_path(projects_path)) + finally: + checkpoint_wal(get_cache_db_path(projects_path)) return result diff --git a/claude_code_log/search.py b/claude_code_log/search.py index 4863e865..9fd29961 100644 --- a/claude_code_log/search.py +++ b/claude_code_log/search.py @@ -732,9 +732,11 @@ def ensure_index( ) for done, file_id in enumerate(todo, start=1): - if file_id in indexed_files: - # Re-indexing a changed file: clear its old rows first. - _delete_file_rows(conn, file_id) + # Clear the file's old rows first — unconditionally, not only when + # the snapshot above knew the file: another process may have + # indexed it since, and a contentless FTS5 table accepts the + # duplicate rowids without a word. A no-op when there are none. + _delete_file_rows(conn, file_id) _index_file(conn, file_id, fields, insert_sql) conn.execute( f"INSERT INTO {FILES_TABLE}" @@ -875,8 +877,11 @@ def catch_up_index(conn: sqlite3.Connection) -> int: conn.execute(f"DELETE FROM {FILES_TABLE} WHERE file_id = ?", (file_id,)) for file_id in todo: with write_transaction(conn): - if file_id in indexed_files: - _delete_file_rows(conn, file_id) + # Inside the transaction and unconditional: the snapshot above + # is stale by the time the lock is held, and a file another + # process indexed meanwhile would otherwise get its rowids + # inserted twice (which a contentless table does not refuse). + _delete_file_rows(conn, file_id) _index_file(conn, file_id, fields, insert_sql) message_count, mtime = current_files[file_id] conn.execute( diff --git a/test/test_search.py b/test/test_search.py index 41db665b..0ad46ba5 100644 --- a/test/test_search.py +++ b/test/test_search.py @@ -1094,6 +1094,113 @@ def record(path: Path) -> int: conn.close() +@pytest.mark.parametrize( + "error", + [ + pytest.param(zlib.error("incorrect header check"), id="zlib"), + pytest.param( + sqlite3.OperationalError("database or disk is full"), id="disk-full" + ), + pytest.param( + sqlite3.DatabaseError("database disk image is malformed"), id="malformed" + ), + ], +) +def test_a_catch_up_error_does_not_fail_the_pass( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + error: Exception, +) -> None: + """The index is derived data that the next pass heals, so an error + while catching it up is a warning: the pass returns, loses no project, + and still checkpoints the WAL.""" + from claude_code_log import converter, search + from claude_code_log.cache import get_cache_db_path + from claude_code_log.converter import process_projects_hierarchy + + projects = tmp_path / "projects" + project = projects / "-home-u-proj" + project.mkdir(parents=True) + jsonl = project / "session.jsonl" + jsonl.write_text(_entry_line("the first version mentions kangaroo")) + process_projects_hierarchy(projects, silent=True) + db_path = get_cache_db_path(projects) + conn = sqlite3.connect(db_path) + ensure_index(conn) + conn.close() + + checkpoints: list[Path] = [] + monkeypatch.setattr( + converter, "checkpoint_wal", lambda path: checkpoints.append(path) + ) + real_catch_up = search.catch_up_index + + def failing(conn: sqlite3.Connection) -> int: + raise error + + monkeypatch.setattr(search, "catch_up_index", failing) + jsonl.write_text(_entry_line("the second version mentions platypus")) + bump_mtime(jsonl) + failed: list[str] = [] + process_projects_hierarchy(projects, silent=True, failed_projects=failed) + + assert failed == [] + assert checkpoints == [db_path], "the WAL checkpoint was skipped" + assert "search index was not brought up to date" in capsys.readouterr().out + + # Nothing changed on disk, but the file's index entry is behind its + # cache row, so the next pass's catch-up heals it. + monkeypatch.setattr(search, "catch_up_index", real_catch_up) + process_projects_hierarchy(projects, silent=True) + conn = sqlite3.connect(db_path) + try: + assert count_matches(conn, "platypus") == 1 + assert count_matches(conn, "kangaroo") == 0 + finally: + conn.close() + + +def test_catch_up_index_tolerates_a_file_indexed_since_its_snapshot( + cache: sqlite3.Connection, monkeypatch: pytest.MonkeyPatch +) -> None: + """A contentless FTS5 table accepts a duplicate rowid without complaint, + so a file another process indexed between the catch-up's snapshot and + its transaction must have its rows cleared before being re-added, or + its tokens count twice.""" + from contextlib import contextmanager + + from claude_code_log import search + + ensure_index(cache) + cache.execute( + "INSERT INTO cached_files(id, project_id, file_name, file_path, " + "source_mtime, cached_mtime, message_count) " + "VALUES(3, 1, 'f3.jsonl', '/tmp/f3.jsonl', 0, 1.0, 1)" + ) + _add_message(cache, 20, 1, 3, _assistant("a brand new file"), uuid="uuid-20") + + real_write_transaction = search.write_transaction + raced = False + + @contextmanager + def racing(conn: sqlite3.Connection): + nonlocal raced + if not raced: + raced = True + # Another process indexes the same new file and commits, after + # our snapshot and before our transaction. + reindex_files(conn, [3]) + with real_write_transaction(conn): + yield + + monkeypatch.setattr(search, "write_transaction", racing) + search.catch_up_index(cache) + + assert raced + assert count_matches(cache, "brand") == 1, "the file's tokens were indexed twice" + + def _entry_line(text: str) -> str: """One user message as a JSONL line.""" return ( From 00524452d1a2e7b4800ec277cf73c742ce6ff402 Mon Sep 17 00:00:00 2001 From: Christian Boos Date: Thu, 5 Nov 2026 09:26:48 +0100 Subject: [PATCH 3/3] Treat an undecodable blob in the catch-up as a warning too The catch-up handler caught sqlite3.Error and zlib.error, but a blob that zlib-decompresses and is not UTF-8 JSON raises from `search.decode_entry` as UnicodeDecodeError or JSONDecodeError, both ValueError: it escaped and failed the pass after the pages were written, with the WAL checkpoint skipped, exactly the case the handler exists for. ValueError joins the tuple. The test puts each kind of blob in the messages table behind a moved cached_mtime, so the real decode path raises inside the pass, and checks the pass returns with no failed project, checkpoints, warns, and that rewriting the file heals the index on the next pass. The other sites that decode a blob are unchanged: `reindex_files` inside an entry write reads back rows the same transaction just serialised, so an undecodable blob there fails that project's save as any error does; `ensure_index` at `serve` start and `search` on the query path were not touched by this branch and keep their behaviour. Co-Authored-By: Claude Fable 5.1 --- claude_code_log/cache.py | 5 +++- test/test_search.py | 63 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 1 deletion(-) diff --git a/claude_code_log/cache.py b/claude_code_log/cache.py index 0535f263..7ddbb689 100644 --- a/claude_code_log/cache.py +++ b/claude_code_log/cache.py @@ -1064,7 +1064,10 @@ def catch_up_search_index(db_path: Path) -> int: return 0 try: return catch_up_index(conn) - except (sqlite3.Error, zlib.error) as exc: + # ValueError covers what `search.decode_entry` raises past zlib: a blob + # that decompresses but is not UTF-8 (UnicodeDecodeError) or not JSON + # (JSONDecodeError) — both subclasses, neither a zlib.error. + except (sqlite3.Error, zlib.error, ValueError) as exc: print( "Warning: the search index was not brought up to date " f"({exc}); the next run or `serve` start will do it." diff --git a/test/test_search.py b/test/test_search.py index 0ad46ba5..1cb15e32 100644 --- a/test/test_search.py +++ b/test/test_search.py @@ -1161,6 +1161,69 @@ def failing(conn: sqlite3.Connection) -> int: conn.close() +@pytest.mark.parametrize( + "blob", + [ + pytest.param(zlib.compress(b"\xff\xfe\x00 not utf-8"), id="not-utf8"), + pytest.param(zlib.compress(b"{not json"), id="not-json"), + ], +) +def test_an_undecodable_blob_in_the_catch_up_does_not_fail_the_pass( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + blob: bytes, +) -> None: + """A blob that decompresses but is not UTF-8 JSON raises a ValueError + (UnicodeDecodeError, JSONDecodeError) from `decode_entry`, not a + zlib.error: the same warning, the pass returns and still checkpoints, + and the next pass that rewrites the file heals the index.""" + from claude_code_log import converter + from claude_code_log.cache import get_cache_db_path + from claude_code_log.converter import process_projects_hierarchy + + projects = tmp_path / "projects" + project = projects / "-home-u-proj" + project.mkdir(parents=True) + jsonl = project / "session.jsonl" + jsonl.write_text(_entry_line("the first version mentions kangaroo")) + process_projects_hierarchy(projects, silent=True) + db_path = get_cache_db_path(projects) + conn = sqlite3.connect(db_path) + ensure_index(conn) + # The file's cache row moved past its index entry (what an entry write + # does), and one of its blobs cannot be decoded. + conn.execute( + "UPDATE messages SET content = ? WHERE id = " + "(SELECT min(id) FROM messages WHERE file_id = 1)", + (blob,), + ) + conn.execute("UPDATE cached_files SET cached_mtime = cached_mtime + 1 WHERE id = 1") + conn.commit() + conn.close() + + checkpoints: list[Path] = [] + monkeypatch.setattr( + converter, "checkpoint_wal", lambda path: checkpoints.append(path) + ) + failed: list[str] = [] + process_projects_hierarchy(projects, silent=True, failed_projects=failed) + + assert failed == [] + assert checkpoints == [db_path], "the WAL checkpoint was skipped" + assert "search index was not brought up to date" in capsys.readouterr().out + + jsonl.write_text(_entry_line("the second version mentions platypus")) + bump_mtime(jsonl) + process_projects_hierarchy(projects, silent=True) + conn = sqlite3.connect(db_path) + try: + assert count_matches(conn, "platypus") == 1 + assert count_matches(conn, "kangaroo") == 0 + finally: + conn.close() + + def test_catch_up_index_tolerates_a_file_indexed_since_its_snapshot( cache: sqlite3.Connection, monkeypatch: pytest.MonkeyPatch ) -> None: