From 1fc58aab42ee1f2787ee92c0a5f93594f74660c2 Mon Sep 17 00:00:00 2001 From: chen Date: Thu, 23 Jul 2026 03:54:02 +0800 Subject: [PATCH 1/8] test: concurrency suite for search-index rebuild, pointer swap, and query --- tests/test_search_index_concurrency.py | 220 +++++++++++++++++++++++++ 1 file changed, 220 insertions(+) create mode 100644 tests/test_search_index_concurrency.py diff --git a/tests/test_search_index_concurrency.py b/tests/test_search_index_concurrency.py new file mode 100644 index 0000000..4343ee8 --- /dev/null +++ b/tests/test_search_index_concurrency.py @@ -0,0 +1,220 @@ +"""Concurrency tests for search-index rebuild, pointer swap, and query.""" + +from __future__ import annotations + +import json +import sqlite3 +import threading +from contextlib import contextmanager +from pathlib import Path +from unittest.mock import patch + +import pytest + +from api.search import _resolve_search_results, _search_via_index +from utils.search_index import build_search_index, query_index_hits, reset_background_for_tests + +_CONCURRENT_ROUNDS = 30 +_READER_THREADS = 4 +_BARRIER_TIMEOUT_S = 45.0 +_ABSENT_TERM = "concurrency-absent-token-xyzzy-999" + + +def _write_session(path: Path, lines: list[dict[str, object]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8") as handle: + for line in lines: + handle.write(json.dumps(line, ensure_ascii=False) + "\n") + + +def _index_patches(cache_root: Path): + return (patch("utils.search_index.cache_dir", return_value=cache_root),) + + +@pytest.fixture +def indexed_tree(tmp_path, monkeypatch): + cache_root = tmp_path / "cache" + cache_root.mkdir() + projects = tmp_path / "projects" + session_path = projects / "demo-proj" / "session_alpha.jsonl" + term = "indexed-unique-sentinel" + _write_session( + session_path, + [ + { + "type": "user", + "timestamp": "2026-05-19T10:00:00Z", + "message": {"content": [{"type": "text", "text": f"find {term}"}]}, + }, + ], + ) + monkeypatch.delenv("CLAUDE_CODE_CHAT_BROWSER_NO_SEARCH_INDEX", raising=False) + monkeypatch.setenv("CLAUDE_CODE_CHAT_BROWSER_SEARCH_INDEX_DIR", str(cache_root)) + reset_background_for_tests() + + patches = _index_patches(cache_root) + with patches[0]: + assert build_search_index(str(projects), [], force=True) is True + yield { + "projects": str(projects), + "cache_root": cache_root, + "term": term, + } + + +def _assert_sentinel_reader_result(term: str, result: dict[str, object]) -> None: + if result["index_locked"]: + return + if not result["query_ok"]: + return + assert result["hits"], ( + "query_ok with zero hits for a unique indexed sentinel during concurrent rebuild" + ) + assert any(term in (hit["text"] or "") for hit in result["hits"]) + + +class TestSearchIndexConcurrency: + def test_concurrent_rebuild_and_query(self, indexed_tree) -> None: + errors: list[BaseException] = [] + barrier = threading.Barrier(_READER_THREADS + 1, timeout=_BARRIER_TIMEOUT_S) + stop = threading.Event() + patches = _index_patches(indexed_tree["cache_root"]) + + def reader() -> None: + try: + barrier.wait(timeout=_BARRIER_TIMEOUT_S) + while not stop.is_set(): + result = query_index_hits( + indexed_tree["term"], + since_ms=None, + max_results=10, + ) + _assert_sentinel_reader_result(indexed_tree["term"], result) + except BaseException as exc: + errors.append(exc) + + def writer() -> None: + try: + barrier.wait(timeout=_BARRIER_TIMEOUT_S) + for _ in range(_CONCURRENT_ROUNDS): + if stop.is_set(): + break + build_search_index(indexed_tree["projects"], [], force=True) + except BaseException as exc: + errors.append(exc) + finally: + stop.set() + + with patches[0]: + threads = [ + threading.Thread(target=reader, name=f"reader-{i}") + for i in range(_READER_THREADS) + ] + threads.append(threading.Thread(target=writer, name="writer")) + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=_BARRIER_TIMEOUT_S + 60.0) + assert not thread.is_alive(), f"{thread.name} did not finish (possible deadlock)" + assert not errors, errors + + def test_operational_error_sets_index_locked(self, indexed_tree) -> None: + @contextmanager + def _locked_conn(*, readonly: bool = True): + class _Conn: + def execute(self, *args: object, **kwargs: object) -> None: + raise sqlite3.OperationalError("database is locked") + + def close(self) -> None: + pass + + yield _Conn() + + patches = _index_patches(indexed_tree["cache_root"]) + with ( + patches[0], + patch("utils.search_index._index_db_conn", _locked_conn), + ): + result = query_index_hits(indexed_tree["term"], since_ms=None, max_results=5) + assert result["index_locked"] is True + assert result["query_ok"] is False + assert result["hits"] == [] + + def test_absent_term_not_confused_with_locked(self, indexed_tree) -> None: + patches = _index_patches(indexed_tree["cache_root"]) + with patches[0]: + result = query_index_hits(_ABSENT_TERM, since_ms=None, max_results=10) + assert result["index_locked"] is False + assert result["query_ok"] is True + assert result["hits"] == [] + + def test_index_locked_without_hits_falls_back_to_live_scan(self, indexed_tree) -> None: + patches = _index_patches(indexed_tree["cache_root"]) + locked_payload = { + "hits": [], + "query_ok": False, + "sql_rows_fetched": 0, + "sql_exhausted": True, + "index_locked": True, + } + with ( + patches[0], + patch("api.search.query_index_hits", return_value=locked_payload), + ): + hits = _resolve_search_results( + indexed_tree["projects"], + [], + indexed_tree["term"], + indexed_tree["term"], + since_ms=None, + max_results=50, + ) + assert len(hits) >= 1 + + def test_search_via_index_returns_partial_hits_when_locked_after_batch(self, indexed_tree) -> None: + patches = _index_patches(indexed_tree["cache_root"]) + fake_hit = { + "session_id": "session_alpha", + "project_name": "demo-proj", + "title": "t", + "role": "user", + "timestamp": "2026-05-19T10:00:00Z", + "text": indexed_tree["term"], + "file_path": "/x", + "mtime": 1.0, + } + calls = {"n": 0} + + def _query_side_effect(*args: object, **kwargs: object) -> dict[str, object]: + calls["n"] += 1 + if calls["n"] == 1: + return { + "hits": [fake_hit], + "query_ok": True, + "sql_rows_fetched": 1, + "sql_exhausted": False, + "index_locked": False, + } + return { + "hits": [], + "query_ok": False, + "sql_rows_fetched": 0, + "sql_exhausted": True, + "index_locked": True, + } + + with ( + patches[0], + patch("api.search.query_index_hits", side_effect=_query_side_effect), + ): + outcome = _search_via_index( + indexed_tree["projects"], + [], + indexed_tree["term"], + indexed_tree["term"], + since_ms=None, + max_results=5, + ) + assert outcome.hits is not None + assert len(outcome.hits) == 1 + assert outcome.index_locked_without_hits is False From b8d7e3954666fc41c859637b095188950d349d4d Mon Sep 17 00:00:00 2001 From: chen Date: Thu, 23 Jul 2026 03:59:09 +0800 Subject: [PATCH 2/8] test: exercise background rebuild and lock order in concurrency suite --- tests/test_search_index_concurrency.py | 182 ++++++++++++++++++++++--- 1 file changed, 160 insertions(+), 22 deletions(-) diff --git a/tests/test_search_index_concurrency.py b/tests/test_search_index_concurrency.py index 4343ee8..0b58470 100644 --- a/tests/test_search_index_concurrency.py +++ b/tests/test_search_index_concurrency.py @@ -2,9 +2,10 @@ from __future__ import annotations -import json import sqlite3 import threading +import time +from collections.abc import Sequence from contextlib import contextmanager from pathlib import Path from unittest.mock import patch @@ -12,23 +13,19 @@ import pytest from api.search import _resolve_search_results, _search_via_index -from utils.search_index import build_search_index, query_index_hits, reset_background_for_tests +from tests.test_search_index import _index_patches, _write_session +from utils.search_index import ( + build_search_index, + query_index_hits, + reset_background_for_tests, + start_search_index_background, +) _CONCURRENT_ROUNDS = 30 _READER_THREADS = 4 _BARRIER_TIMEOUT_S = 45.0 _ABSENT_TERM = "concurrency-absent-token-xyzzy-999" - - -def _write_session(path: Path, lines: list[dict[str, object]]) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - with path.open("w", encoding="utf-8") as handle: - for line in lines: - handle.write(json.dumps(line, ensure_ascii=False) + "\n") - - -def _index_patches(cache_root: Path): - return (patch("utils.search_index.cache_dir", return_value=cache_root),) +_BACKGROUND_POLL_S = 1 @pytest.fixture @@ -73,17 +70,85 @@ def _assert_sentinel_reader_result(term: str, result: dict[str, object]) -> None assert any(term in (hit["text"] or "") for hit in result["hits"]) +def _lock_events_satisfy_documented_contract( + events: Sequence[tuple[str, str]], +) -> bool: + """True when acquisition order matches docs/architecture.md lock contract.""" + held: set[str] = set() + build_depth = 0 + for name, action in events: + if action == "acquire": + if name == "index": + return False + if name == "usability" and "build" not in held: + return False + held.add(name) + if name == "build": + build_depth += 1 + elif action == "release": + if name not in held: + return False + held.remove(name) + if name == "build": + build_depth -= 1 + return build_depth == 0 and not held + + +def _record_lock_events_during_build( + cache_root, + projects: str, +) -> list[tuple[str, str]]: + import utils.search_index as si + + events: list[tuple[str, str]] = [] + + class _TrackedLock: + def __init__(self, name: str) -> None: + self._name = name + self._inner = threading.Lock() + + def acquire(self, blocking: bool = True, timeout: float = -1) -> bool: + events.append((self._name, "acquire")) + return self._inner.acquire(blocking, timeout) + + def release(self) -> None: + events.append((self._name, "release")) + self._inner.release() + + def __enter__(self) -> _TrackedLock: + self.acquire() + return self + + def __exit__(self, *args: object) -> None: + self.release() + + saved_build = si._index_build_lock + saved_usability = si._usability_cache_lock + saved_index = si._index_lock + si._index_build_lock = _TrackedLock("build") + si._usability_cache_lock = _TrackedLock("usability") + si._index_lock = _TrackedLock("index") + patches = _index_patches(cache_root) + try: + with patches[0]: + build_search_index(projects, [], force=True) + finally: + si._index_build_lock = saved_build + si._usability_cache_lock = saved_usability + si._index_lock = saved_index + return events + + class TestSearchIndexConcurrency: def test_concurrent_rebuild_and_query(self, indexed_tree) -> None: errors: list[BaseException] = [] barrier = threading.Barrier(_READER_THREADS + 1, timeout=_BARRIER_TIMEOUT_S) - stop = threading.Event() patches = _index_patches(indexed_tree["cache_root"]) def reader() -> None: try: - barrier.wait(timeout=_BARRIER_TIMEOUT_S) - while not stop.is_set(): + for _ in range(_CONCURRENT_ROUNDS): + barrier.wait(timeout=_BARRIER_TIMEOUT_S) result = query_index_hits( indexed_tree["term"], since_ms=None, @@ -95,15 +160,11 @@ def reader() -> None: def writer() -> None: try: - barrier.wait(timeout=_BARRIER_TIMEOUT_S) for _ in range(_CONCURRENT_ROUNDS): - if stop.is_set(): - break + barrier.wait(timeout=_BARRIER_TIMEOUT_S) build_search_index(indexed_tree["projects"], [], force=True) except BaseException as exc: errors.append(exc) - finally: - stop.set() with patches[0]: threads = [ @@ -114,10 +175,87 @@ def writer() -> None: for thread in threads: thread.start() for thread in threads: - thread.join(timeout=_BARRIER_TIMEOUT_S + 60.0) + thread.join(timeout=_BARRIER_TIMEOUT_S + 90.0) assert not thread.is_alive(), f"{thread.name} did not finish (possible deadlock)" assert not errors, errors + def test_queries_during_background_refresh(self, indexed_tree) -> None: + errors: list[BaseException] = [] + stop = threading.Event() + projects = indexed_tree["projects"] + patches = _index_patches(indexed_tree["cache_root"]) + + def reader() -> None: + try: + while not stop.is_set(): + result = query_index_hits( + indexed_tree["term"], + since_ms=None, + max_results=10, + ) + _assert_sentinel_reader_result(indexed_tree["term"], result) + time.sleep(0.02) + except BaseException as exc: + errors.append(exc) + + def mutator() -> None: + try: + for i in range(5): + if stop.is_set(): + break + session_path = ( + Path(projects) / "demo-proj" / f"session_bg_{i}.jsonl" + ) + _write_session( + session_path, + [ + { + "type": "user", + "timestamp": "2026-05-19T10:00:00Z", + "message": { + "content": [ + { + "type": "text", + "text": f"find {indexed_tree['term']} bg{i}", + } + ] + }, + }, + ], + ) + time.sleep(0.15) + except BaseException as exc: + errors.append(exc) + finally: + stop.set() + + with patches[0]: + reset_background_for_tests() + start_search_index_background(projects, [], poll_seconds=_BACKGROUND_POLL_S) + threads = [ + threading.Thread(target=reader, name="reader"), + threading.Thread(target=mutator, name="mutator"), + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=8.0) + assert not thread.is_alive() + reset_background_for_tests() + assert not errors, errors + + def test_documented_lock_order_during_build(self, indexed_tree) -> None: + events = _record_lock_events_during_build( + indexed_tree["cache_root"], + indexed_tree["projects"], + ) + assert events, "expected lock events during build_search_index" + assert _lock_events_satisfy_documented_contract(events) + + def test_lock_order_invariant_rejects_forbidden_nesting(self) -> None: + forbidden = [("usability", "acquire"), ("build", "acquire")] + assert not _lock_events_satisfy_documented_contract(forbidden) + def test_operational_error_sets_index_locked(self, indexed_tree) -> None: @contextmanager def _locked_conn(*, readonly: bool = True): From e20e37b9c33bcec0dbde5f4dbf106840afedcd2d Mon Sep 17 00:00:00 2001 From: chen Date: Thu, 23 Jul 2026 04:53:50 +0800 Subject: [PATCH 3/8] test: fix ruff and mypy on search-index concurrency tests --- tests/test_search_index_concurrency.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/tests/test_search_index_concurrency.py b/tests/test_search_index_concurrency.py index 0b58470..0ad192d 100644 --- a/tests/test_search_index_concurrency.py +++ b/tests/test_search_index_concurrency.py @@ -15,6 +15,7 @@ from api.search import _resolve_search_results, _search_via_index from tests.test_search_index import _index_patches, _write_session from utils.search_index import ( + IndexQueryResult, build_search_index, query_index_hits, reset_background_for_tests, @@ -59,7 +60,7 @@ def indexed_tree(tmp_path, monkeypatch): } -def _assert_sentinel_reader_result(term: str, result: dict[str, object]) -> None: +def _assert_sentinel_reader_result(term: str, result: IndexQueryResult) -> None: if result["index_locked"]: return if not result["query_ok"]: @@ -95,7 +96,7 @@ def _lock_events_satisfy_documented_contract( def _record_lock_events_during_build( - cache_root, + cache_root: Path, projects: str, ) -> list[tuple[str, str]]: import utils.search_index as si @@ -168,8 +169,7 @@ def writer() -> None: with patches[0]: threads = [ - threading.Thread(target=reader, name=f"reader-{i}") - for i in range(_READER_THREADS) + threading.Thread(target=reader, name=f"reader-{i}") for i in range(_READER_THREADS) ] threads.append(threading.Thread(target=writer, name="writer")) for thread in threads: @@ -203,9 +203,7 @@ def mutator() -> None: for i in range(5): if stop.is_set(): break - session_path = ( - Path(projects) / "demo-proj" / f"session_bg_{i}.jsonl" - ) + session_path = Path(projects) / "demo-proj" / f"session_bg_{i}.jsonl" _write_session( session_path, [ @@ -309,7 +307,9 @@ def test_index_locked_without_hits_falls_back_to_live_scan(self, indexed_tree) - ) assert len(hits) >= 1 - def test_search_via_index_returns_partial_hits_when_locked_after_batch(self, indexed_tree) -> None: + def test_search_via_index_returns_partial_hits_when_locked_after_batch( + self, indexed_tree + ) -> None: patches = _index_patches(indexed_tree["cache_root"]) fake_hit = { "session_id": "session_alpha", From 44a8f29e0a4944c8ed57f5cef56fb973cc50239a Mon Sep 17 00:00:00 2001 From: chen Date: Thu, 23 Jul 2026 06:29:54 +0800 Subject: [PATCH 4/8] fix: stoppable background worker and stronger concurrency test oracles --- tests/conftest.py | 16 ++++ tests/test_search_index.py | 17 ++-- tests/test_search_index_concurrency.py | 111 ++++++++++++++++++------- utils/search_index.py | 109 ++++++++++++------------ 4 files changed, 159 insertions(+), 94 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 5c08244..5f75f00 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,16 +2,32 @@ from __future__ import annotations +import json import os import shutil from collections.abc import Mapping from pathlib import Path +from unittest.mock import patch import pytest from hypothesis import settings from app import create_app + +def write_session(path: Path, lines: list[dict[str, object]]) -> None: + """Write JSONL session *lines* to *path*, creating parent directories.""" + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8") as handle: + for line in lines: + handle.write(json.dumps(line, ensure_ascii=False) + "\n") + + +def index_patches(cache_root: Path): + """Return the patch context managers that point the index at *cache_root*.""" + return (patch("utils.search_index.cache_dir", return_value=cache_root),) + + # Hypothesis profiles drive fuzz example counts/deadlines (deadline disabled to # avoid timing flakiness on slow/CI runners). CI runs fewer examples for speed. settings.register_profile("dev", max_examples=200, deadline=None) diff --git a/tests/test_search_index.py b/tests/test_search_index.py index 239af2e..06d5c16 100644 --- a/tests/test_search_index.py +++ b/tests/test_search_index.py @@ -2,7 +2,6 @@ from __future__ import annotations -import json import sqlite3 from contextlib import closing, contextmanager from datetime import UTC, datetime, timedelta @@ -12,6 +11,7 @@ import pytest from api.search import _resolve_search_results, _search_via_index +from tests.conftest import index_patches as _index_patches, write_session as _write_session from utils.exclusion_rules import load_rules from utils.search_index import ( build_search_index, @@ -26,17 +26,6 @@ ) -def _write_session(path: Path, lines: list[dict[str, object]]) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - with path.open("w", encoding="utf-8") as handle: - for line in lines: - handle.write(json.dumps(line, ensure_ascii=False) + "\n") - - -def _index_patches(cache_root: Path): - return (patch("utils.search_index.cache_dir", return_value=cache_root),) - - @pytest.fixture def indexed_tree(tmp_path, monkeypatch): """Temp projects dir + isolated search index cache.""" @@ -381,11 +370,15 @@ def test_background_worker_starts_once(self, indexed_tree, monkeypatch): patches[0], patch("utils.search_index.threading.Thread") as mock_thread, ): + # The mock worker is not a real thread; report it as stopped so the + # teardown reset does not treat it as a live worker that won't join. + mock_thread.return_value.is_alive.return_value = False start_search_index_background(indexed_tree["projects"], []) start_search_index_background(indexed_tree["projects"], []) assert mock_thread.call_count == 1 assert mock_thread.call_args.kwargs.get("daemon") is True assert mock_thread.return_value.start.call_count == 1 + reset_background_for_tests() class TestIndexSearchCompleteness: diff --git a/tests/test_search_index_concurrency.py b/tests/test_search_index_concurrency.py index 0ad192d..6e481bc 100644 --- a/tests/test_search_index_concurrency.py +++ b/tests/test_search_index_concurrency.py @@ -13,10 +13,11 @@ import pytest from api.search import _resolve_search_results, _search_via_index -from tests.test_search_index import _index_patches, _write_session +from tests.conftest import index_patches as _index_patches, write_session as _write_session from utils.search_index import ( IndexQueryResult, build_search_index, + ensure_search_index, query_index_hits, reset_background_for_tests, start_search_index_background, @@ -27,6 +28,28 @@ _BARRIER_TIMEOUT_S = 45.0 _ABSENT_TERM = "concurrency-absent-token-xyzzy-999" _BACKGROUND_POLL_S = 1 +_BACKGROUND_MUTATOR_ITERATIONS = 12 +_BACKGROUND_MUTATOR_SLEEP_S = 0.25 +_BACKGROUND_THREAD_JOIN_TIMEOUT_S = 12.0 +_BACKGROUND_REFRESH_WAIT_S = 8.0 +_MIN_BACKGROUND_REFRESHES = 2 + +# A locked index reports no hits with query_ok False; used to drive the +# live-scan fallback paths in api.search. +_LOCKED_PAYLOAD: IndexQueryResult = { + "hits": [], + "query_ok": False, + "sql_rows_fetched": 0, + "sql_exhausted": True, + "index_locked": True, +} + + +@pytest.fixture(autouse=True) +def _isolate_search_index_background_worker(): + reset_background_for_tests() + yield + reset_background_for_tests() @pytest.fixture @@ -185,6 +208,15 @@ def test_queries_during_background_refresh(self, indexed_tree) -> None: projects = indexed_tree["projects"] patches = _index_patches(indexed_tree["cache_root"]) + refresh_lock = threading.Lock() + refresh_count = 0 + + def _counting_ensure(*args: object, **kwargs: object) -> None: + nonlocal refresh_count + with refresh_lock: + refresh_count += 1 + ensure_search_index(*args, **kwargs) # type: ignore[arg-type] + def reader() -> None: try: while not stop.is_set(): @@ -200,7 +232,7 @@ def reader() -> None: def mutator() -> None: try: - for i in range(5): + for i in range(_BACKGROUND_MUTATOR_ITERATIONS): if stop.is_set(): break session_path = Path(projects) / "demo-proj" / f"session_bg_{i}.jsonl" @@ -221,25 +253,57 @@ def mutator() -> None: }, ], ) - time.sleep(0.15) + time.sleep(_BACKGROUND_MUTATOR_SLEEP_S) except BaseException as exc: errors.append(exc) finally: stop.set() + def _bg_term_visible() -> bool: + # A term only present in a mutator-written session becomes queryable + # once the background worker rebuilds the index and swaps the pointer. + result = query_index_hits( + f"{indexed_tree['term']} bg0", + since_ms=None, + max_results=10, + ) + return result["query_ok"] and bool(result["hits"]) + with patches[0]: - reset_background_for_tests() - start_search_index_background(projects, [], poll_seconds=_BACKGROUND_POLL_S) - threads = [ - threading.Thread(target=reader, name="reader"), - threading.Thread(target=mutator, name="mutator"), - ] - for thread in threads: - thread.start() - for thread in threads: - thread.join(timeout=8.0) - assert not thread.is_alive() - reset_background_for_tests() + try: + reset_background_for_tests() + with patch( + "utils.search_index.ensure_search_index", + side_effect=_counting_ensure, + ): + start_search_index_background(projects, [], poll_seconds=_BACKGROUND_POLL_S) + threads = [ + threading.Thread(target=reader, name="reader"), + threading.Thread(target=mutator, name="mutator"), + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=_BACKGROUND_THREAD_JOIN_TIMEOUT_S) + assert not thread.is_alive() + + deadline = time.monotonic() + _BACKGROUND_REFRESH_WAIT_S + bg_visible = False + while time.monotonic() < deadline: + if _bg_term_visible(): + bg_visible = True + break + time.sleep(0.05) + + assert bg_visible, "background worker never rebuilt index with new sessions" + with refresh_lock: + observed = refresh_count + assert observed >= _MIN_BACKGROUND_REFRESHES, ( + f"expected at least {_MIN_BACKGROUND_REFRESHES} background " + f"refreshes, observed {observed}" + ) + finally: + reset_background_for_tests() assert not errors, errors def test_documented_lock_order_during_build(self, indexed_tree) -> None: @@ -286,16 +350,9 @@ def test_absent_term_not_confused_with_locked(self, indexed_tree) -> None: def test_index_locked_without_hits_falls_back_to_live_scan(self, indexed_tree) -> None: patches = _index_patches(indexed_tree["cache_root"]) - locked_payload = { - "hits": [], - "query_ok": False, - "sql_rows_fetched": 0, - "sql_exhausted": True, - "index_locked": True, - } with ( patches[0], - patch("api.search.query_index_hits", return_value=locked_payload), + patch("api.search.query_index_hits", return_value=_LOCKED_PAYLOAD), ): hits = _resolve_search_results( indexed_tree["projects"], @@ -333,13 +390,7 @@ def _query_side_effect(*args: object, **kwargs: object) -> dict[str, object]: "sql_exhausted": False, "index_locked": False, } - return { - "hits": [], - "query_ok": False, - "sql_rows_fetched": 0, - "sql_exhausted": True, - "index_locked": True, - } + return dict(_LOCKED_PAYLOAD) with ( patches[0], diff --git a/utils/search_index.py b/utils/search_index.py index a09beae..8c7d79b 100644 --- a/utils/search_index.py +++ b/utils/search_index.py @@ -58,10 +58,15 @@ _index_build_lock = threading.Lock() _background_started = False _background_lock_fd: int | None = None +_background_stop = threading.Event() +_background_thread: threading.Thread | None = None _usability_cache: dict[tuple[str, str], tuple[bool, float]] = {} _usability_cache_lock = threading.Lock() _USABILITY_CACHE_TTL_SECONDS = 30.0 _FTS_BATCH_SIZE = 200 +_DEFAULT_TITLE = "Untitled Session" +_ACTIVE_POINTER_NAME = "search_index.active" +_BACKGROUND_JOIN_TIMEOUT_S = 10.0 class IndexMessageHitDict(TypedDict): @@ -83,6 +88,17 @@ class IndexQueryResult(TypedDict): index_locked: bool +def _empty_query_result(*, index_locked: bool = False) -> IndexQueryResult: + """Return a no-hits query result; ``index_locked`` distinguishes lock from empty.""" + return { + "hits": [], + "query_ok": False, + "sql_rows_fetched": 0, + "sql_exhausted": True, + "index_locked": index_locked, + } + + def cache_dir() -> Path: """Return directory for search index files (overridable in tests).""" override = os.environ.get("CLAUDE_CODE_CHAT_BROWSER_SEARCH_INDEX_DIR", "").strip() @@ -168,7 +184,7 @@ def search_snippet(text: str, query: str, *, context: int = 80) -> str: def _resolve_active_index_db_path() -> Path | None: - pointer = cache_dir() / "search_index.active" + pointer = cache_dir() / _ACTIVE_POINTER_NAME legacy = cache_dir() / "search_index.sqlite" if pointer.is_file(): try: @@ -187,7 +203,7 @@ def _resolve_active_index_db_path() -> Path | None: def _publish_active_index(new_db_path: Path) -> None: root = cache_dir() root.mkdir(parents=True, exist_ok=True) - pointer = root / "search_index.active" + pointer = root / _ACTIVE_POINTER_NAME pointer_tmp = pointer.with_suffix(".active.tmp") pointer_tmp.write_text(new_db_path.name, encoding="utf-8") try: @@ -465,7 +481,7 @@ def _scan_session_file( filepath: str, ) -> tuple[str, str, int, int, float, list[tuple[str, str | None, RoleLiteral]]]: session_id = os.path.basename(filepath).replace(".jsonl", "") - title = "Untitled Session" + title = _DEFAULT_TITLE first_ms = 0 last_ms = 0 messages: list[tuple[str, str | None, RoleLiteral]] = [] @@ -491,7 +507,7 @@ def _scan_session_file( first_ms = ms last_ms = ms - if title == "Untitled Session" and entry.get("type") == "user": + if title == _DEFAULT_TITLE and entry.get("type") == "user": msg = entry_message(entry) text = extract_text(msg.get("content", [])) if text: @@ -670,25 +686,28 @@ def start_search_index_background( poll_seconds: int = 60, ) -> None: """Kick off initial + periodic index refresh in a daemon thread.""" - global _background_started + global _background_started, _background_thread if not index_search_enabled(): return - with _index_lock: - if _background_started: - return - if not _try_acquire_cross_process_background_lock(): - return - _background_started = True def _worker() -> None: - while True: + while not _background_stop.is_set(): try: ensure_search_index(projects_dir, rules) except Exception: _logger.exception("Background search index refresh failed") - time.sleep(poll_seconds) + if _background_stop.wait(timeout=poll_seconds): + break - thread = threading.Thread(target=_worker, name="search-index-refresh", daemon=True) + with _index_lock: + if _background_started: + return + if not _try_acquire_cross_process_background_lock(): + return + _background_started = True + _background_stop.clear() + thread = threading.Thread(target=_worker, name="search-index-refresh", daemon=True) + _background_thread = thread thread.start() @@ -726,34 +745,16 @@ def query_index_hits( ) -> IndexQueryResult: """Return message hits from the FTS index (pre-exclusion, pre-snippet).""" if not query_lower or not index_search_enabled(): - return { - "hits": [], - "query_ok": False, - "sql_rows_fetched": 0, - "sql_exhausted": True, - "index_locked": False, - } + return _empty_query_result() fts_q = _fts_match_query(query_lower) if not fts_q: - return { - "hits": [], - "query_ok": False, - "sql_rows_fetched": 0, - "sql_exhausted": True, - "index_locked": False, - } + return _empty_query_result() sql_limit = max(max_results, _FTS_BATCH_SIZE) with _index_db_conn(readonly=True) as conn: if conn is None: - return { - "hits": [], - "query_ok": False, - "sql_rows_fetched": 0, - "sql_exhausted": True, - "index_locked": False, - } + return _empty_query_result() try: rows = conn.execute( "SELECT m.session_id, m.project_name, m.role, m.timestamp_ms, m.text," @@ -768,22 +769,10 @@ def query_index_hits( ).fetchall() except sqlite3.OperationalError as exc: _logger.debug("FTS query locked (%s); index may be rebuilding", exc) - return { - "hits": [], - "query_ok": False, - "sql_rows_fetched": 0, - "sql_exhausted": True, - "index_locked": True, - } + return _empty_query_result(index_locked=True) except sqlite3.Error as exc: _logger.debug("FTS query failed (%s); index may be rebuilding", exc) - return { - "hits": [], - "query_ok": False, - "sql_rows_fetched": 0, - "sql_exhausted": True, - "index_locked": False, - } + return _empty_query_result() hits: list[IndexMessageHitDict] = [] rows_scanned = 0 @@ -799,7 +788,7 @@ def query_index_hits( IndexMessageHitDict( session_id=row["session_id"], project_name=row["project_name"], - title=row["title"] or "Untitled Session", + title=row["title"] or _DEFAULT_TITLE, role=row["role"], timestamp=ms_to_timestamp(ts_ms), text=text, @@ -820,14 +809,30 @@ def query_index_hits( def reset_background_for_tests() -> None: - """Allow tests to restart the background worker.""" - global _background_started, _background_lock_fd + """Stop the background worker and allow tests to restart it.""" + global _background_started, _background_lock_fd, _background_thread + _background_stop.set() + thread = _background_thread + worker_stopped = True + if thread is not None and thread.is_alive(): + thread.join(timeout=_BACKGROUND_JOIN_TIMEOUT_S) + worker_stopped = not thread.is_alive() + if not worker_stopped: + _logger.warning( + "background search-index worker did not stop within %.0fs", + _BACKGROUND_JOIN_TIMEOUT_S, + ) with _index_lock: _background_started = False + _background_thread = None if _background_lock_fd is not None: try: os.close(_background_lock_fd) except OSError: pass _background_lock_fd = None + # Leave the stop event set when the worker never terminated so it exits on + # its next loop check and no second worker races it; start() re-clears it. + if worker_stopped: + _background_stop.clear() _clear_usability_cache() From 93e0b4229749bcc9a710d621991c177afe9739db Mon Sep 17 00:00:00 2001 From: chen Date: Thu, 23 Jul 2026 06:40:03 +0800 Subject: [PATCH 5/8] fix: keep background worker ownership until stop is confirmed --- utils/search_index.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/utils/search_index.py b/utils/search_index.py index 8c7d79b..f4ce0a6 100644 --- a/utils/search_index.py +++ b/utils/search_index.py @@ -708,7 +708,7 @@ def _worker() -> None: _background_stop.clear() thread = threading.Thread(target=_worker, name="search-index-refresh", daemon=True) _background_thread = thread - thread.start() + thread.start() def index_is_usable(projects_dir: str, rules: list[Any]) -> bool: @@ -822,6 +822,9 @@ def reset_background_for_tests() -> None: "background search-index worker did not stop within %.0fs", _BACKGROUND_JOIN_TIMEOUT_S, ) + if not worker_stopped: + _clear_usability_cache() + return with _index_lock: _background_started = False _background_thread = None @@ -831,8 +834,5 @@ def reset_background_for_tests() -> None: except OSError: pass _background_lock_fd = None - # Leave the stop event set when the worker never terminated so it exits on - # its next loop check and no second worker races it; start() re-clears it. - if worker_stopped: - _background_stop.clear() + _background_stop.clear() _clear_usability_cache() From ac062d17b0cebe2ffcc61e63d12ea671a99dd950 Mon Sep 17 00:00:00 2001 From: chen Date: Thu, 23 Jul 2026 18:37:08 +0800 Subject: [PATCH 6/8] address per comment --- tests/test_search_index_concurrency.py | 17 +++++++++++++++++ utils/search_index.py | 7 +------ 2 files changed, 18 insertions(+), 6 deletions(-) diff --git a/tests/test_search_index_concurrency.py b/tests/test_search_index_concurrency.py index 6e481bc..f70daaa 100644 --- a/tests/test_search_index_concurrency.py +++ b/tests/test_search_index_concurrency.py @@ -83,6 +83,14 @@ def indexed_tree(tmp_path, monkeypatch): } +def _sentinel_hit_observed(term: str, result: IndexQueryResult) -> bool: + if result["index_locked"] or not result["query_ok"]: + return False + if not result["hits"]: + return False + return any(term in (hit["text"] or "") for hit in result["hits"]) + + def _assert_sentinel_reader_result(term: str, result: IndexQueryResult) -> None: if result["index_locked"]: return @@ -168,8 +176,11 @@ def test_concurrent_rebuild_and_query(self, indexed_tree) -> None: errors: list[BaseException] = [] barrier = threading.Barrier(_READER_THREADS + 1, timeout=_BARRIER_TIMEOUT_S) patches = _index_patches(indexed_tree["cache_root"]) + hits_lock = threading.Lock() + sentinel_hits_observed = 0 def reader() -> None: + nonlocal sentinel_hits_observed try: for _ in range(_CONCURRENT_ROUNDS): barrier.wait(timeout=_BARRIER_TIMEOUT_S) @@ -179,6 +190,9 @@ def reader() -> None: max_results=10, ) _assert_sentinel_reader_result(indexed_tree["term"], result) + if _sentinel_hit_observed(indexed_tree["term"], result): + with hits_lock: + sentinel_hits_observed += 1 except BaseException as exc: errors.append(exc) @@ -201,6 +215,9 @@ def writer() -> None: thread.join(timeout=_BARRIER_TIMEOUT_S + 90.0) assert not thread.is_alive(), f"{thread.name} did not finish (possible deadlock)" assert not errors, errors + assert sentinel_hits_observed > 0, ( + "no reader round observed sentinel hits during concurrent rebuild" + ) def test_queries_during_background_refresh(self, indexed_tree) -> None: errors: list[BaseException] = [] diff --git a/utils/search_index.py b/utils/search_index.py index f4ce0a6..43f5ec8 100644 --- a/utils/search_index.py +++ b/utils/search_index.py @@ -813,18 +813,13 @@ def reset_background_for_tests() -> None: global _background_started, _background_lock_fd, _background_thread _background_stop.set() thread = _background_thread - worker_stopped = True if thread is not None and thread.is_alive(): thread.join(timeout=_BACKGROUND_JOIN_TIMEOUT_S) - worker_stopped = not thread.is_alive() - if not worker_stopped: + if thread.is_alive(): _logger.warning( "background search-index worker did not stop within %.0fs", _BACKGROUND_JOIN_TIMEOUT_S, ) - if not worker_stopped: - _clear_usability_cache() - return with _index_lock: _background_started = False _background_thread = None From f726eb521b12ccc9dac2696854cc1141da224577 Mon Sep 17 00:00:00 2001 From: chen Date: Thu, 23 Jul 2026 23:11:49 +0800 Subject: [PATCH 7/8] fix docstring --- tests/conftest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/conftest.py b/tests/conftest.py index 5f75f00..0848a52 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -83,7 +83,7 @@ def client(tmp_path, export_state_file): @pytest.fixture def client_single(tmp_path, export_state_file): - """Flask test client with one seeded session ? for search/limit tests.""" + """Flask test client with one seeded session for search/limit tests.""" return _make_test_client(tmp_path, {"session_abc123.jsonl": "session_minimal.jsonl"}) From 811cf7e99c4a657700236a434bfb7d8c1df58e2a Mon Sep 17 00:00:00 2001 From: chen Date: Fri, 24 Jul 2026 00:02:42 +0800 Subject: [PATCH 8/8] Pin locked-index live-scan fallback in concurrency test --- tests/test_search_index_concurrency.py | 36 ++++++++++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/tests/test_search_index_concurrency.py b/tests/test_search_index_concurrency.py index f70daaa..d5daaeb 100644 --- a/tests/test_search_index_concurrency.py +++ b/tests/test_search_index_concurrency.py @@ -12,7 +12,12 @@ import pytest -from api.search import _resolve_search_results, _search_via_index +from api.search import ( + _live_scan_with_index_lock_fallback, + _merge_search_hits, + _resolve_search_results, + _search_via_index, +) from tests.conftest import index_patches as _index_patches, write_session as _write_session from utils.search_index import ( IndexQueryResult, @@ -371,7 +376,7 @@ def test_index_locked_without_hits_falls_back_to_live_scan(self, indexed_tree) - patches[0], patch("api.search.query_index_hits", return_value=_LOCKED_PAYLOAD), ): - hits = _resolve_search_results( + outcome = _search_via_index( indexed_tree["projects"], [], indexed_tree["term"], @@ -379,6 +384,33 @@ def test_index_locked_without_hits_falls_back_to_live_scan(self, indexed_tree) - since_ms=None, max_results=50, ) + assert outcome.hits is None + assert outcome.index_locked_without_hits is True + + lock_fallback_flags: list[bool] = [] + + def _record_lock_fallback(*args: object, **kwargs: object) -> list[object]: + lock_fallback_flags.append(bool(kwargs["index_locked_without_hits"])) + return _live_scan_with_index_lock_fallback(*args, **kwargs) + + with ( + patch( + "api.search._live_scan_with_index_lock_fallback", + side_effect=_record_lock_fallback, + ), + patch("api.search._merge_search_hits", wraps=_merge_search_hits) as merge_mock, + ): + hits = _resolve_search_results( + indexed_tree["projects"], + [], + indexed_tree["term"], + indexed_tree["term"], + since_ms=None, + max_results=50, + ) + + assert lock_fallback_flags == [True] + merge_mock.assert_not_called() assert len(hits) >= 1 def test_search_via_index_returns_partial_hits_when_locked_after_batch(