(Claude) Maintain the search index after the pass, not under the write lock

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

claude_code_log/cache.py

@@ -1030,6 +1030,54 @@ 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`).
+
+    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
+
+    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)
+    # 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."
+        )
+        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 +1759,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 +1904,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)

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,33 +6596,45 @@ 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).
-            checkpoint_wal(get_cache_db_path(projects_path))
+            # 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). 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
 
 
@@ -7069,7 +7094,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 = {

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
@@ -730,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}"
@@ -779,6 +783,121 @@ 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):
+            # 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(
+                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 +912,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.