diff --git a/tests/conftest.py b/tests/conftest.py index 5c08244..0848a52 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) @@ -67,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"}) 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 new file mode 100644 index 0000000..d5daaeb --- /dev/null +++ b/tests/test_search_index_concurrency.py @@ -0,0 +1,458 @@ +"""Concurrency tests for search-index rebuild, pointer swap, and query.""" + +from __future__ import annotations + +import sqlite3 +import threading +import time +from collections.abc import Sequence +from contextlib import contextmanager +from pathlib import Path +from unittest.mock import patch + +import pytest + +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, + build_search_index, + ensure_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" +_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 +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 _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 + 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"]) + + +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: Path, + 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) + 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) + result = query_index_hits( + indexed_tree["term"], + since_ms=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) + + def writer() -> None: + try: + for _ in range(_CONCURRENT_ROUNDS): + barrier.wait(timeout=_BARRIER_TIMEOUT_S) + build_search_index(indexed_tree["projects"], [], force=True) + except BaseException as exc: + errors.append(exc) + + 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 + 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] = [] + stop = threading.Event() + 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(): + 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(_BACKGROUND_MUTATOR_ITERATIONS): + 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(_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]: + 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: + 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): + 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"]) + with ( + patches[0], + patch("api.search.query_index_hits", return_value=_LOCKED_PAYLOAD), + ): + outcome = _search_via_index( + indexed_tree["projects"], + [], + indexed_tree["term"], + indexed_tree["term"], + 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( + 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 dict(_LOCKED_PAYLOAD) + + 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 diff --git a/utils/search_index.py b/utils/search_index.py index a09beae..43f5ec8 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,26 +686,29 @@ 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) - thread.start() + 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() def index_is_usable(projects_dir: str, rules: list[Any]) -> bool: @@ -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,25 @@ 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 + if thread is not None and thread.is_alive(): + thread.join(timeout=_BACKGROUND_JOIN_TIMEOUT_S) + if thread.is_alive(): + _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 + _background_stop.clear() _clear_usability_cache()