feat: add SEARCH_INDEX_UNAVAILABLE when locked index fallback fails

Author: clean6378-max-it

api/search.py

@@ -5,7 +5,7 @@
 import logging
 import os
 from datetime import datetime, timezone
-from typing import Any
+from typing import Any, NamedTuple
 
 from flask import Blueprint, current_app, request
 
@@ -36,6 +36,16 @@
 _MAX_SEARCH_SINCE_DAYS = 36_500
 
 
+class _IndexSearchOutcome(NamedTuple):
+    hits: list[SearchHitDict] | None
+    fts_exhausted: bool
+    index_locked_without_hits: bool = False
+
+
+class _SearchIndexUnavailableError(Exception):
+    """Index locked with no partial hits and live-scan fallback failed."""
+
+
 def _parse_limit(raw: str | None, default: int = _DEFAULT_LIMIT) -> int:
     """Parse a positive integer limit from a query string value."""
     if raw is None or raw.strip() == "":
@@ -145,9 +155,9 @@ def _search_via_index(
     *,
     since_ms: int | None,
     max_results: int,
-) -> tuple[list[SearchHitDict] | None, bool]:
+) -> _IndexSearchOutcome:
     if not index_search_enabled() or not index_is_usable(projects_dir, rules):
-        return None, False
+        return _IndexSearchOutcome(None, False)
 
     rules_fp = rules_fingerprint(rules)
     results: list[SearchHitDict] = []
@@ -167,10 +177,10 @@ def _search_via_index(
                 len(results),
             )
             if results:
-                return _rank_search_hits(results)[:max_results], False
-            return None, False
+                return _IndexSearchOutcome(_rank_search_hits(results)[:max_results], False)
+            return _IndexSearchOutcome(None, False, index_locked_without_hits=True)
         if not indexed["query_ok"]:
-            return None, False
+            return _IndexSearchOutcome(None, False)
         if indexed["sql_rows_fetched"] == 0:
             fts_exhausted = indexed["sql_exhausted"]
             break
@@ -201,7 +211,36 @@ def _search_via_index(
         if indexed["sql_exhausted"]:
             fts_exhausted = True
             break
-    return _rank_search_hits(results)[:max_results], fts_exhausted
+    return _IndexSearchOutcome(_rank_search_hits(results)[:max_results], fts_exhausted)
+
+
+def _live_scan_with_index_lock_fallback(
+    base: str,
+    rules: list[Any],
+    query: str,
+    query_lower: str,
+    *,
+    since_ms: int | None,
+    max_results: int,
+    index_locked_without_hits: bool,
+) -> list[SearchHitDict]:
+    try:
+        return _search_live_scan(
+            base,
+            rules,
+            query,
+            query_lower,
+            since_ms=since_ms,
+            max_results=max_results,
+        )
+    except Exception:
+        if index_locked_without_hits:
+            _logger.warning(
+                "Search index locked; live-scan fallback failed",
+                exc_info=True,
+            )
+            raise _SearchIndexUnavailableError from None
+        raise
 
 
 def _resolve_search_results(
@@ -213,36 +252,38 @@ def _resolve_search_results(
     since_ms: int | None,
     max_results: int,
 ) -> list[SearchHitDict]:
-    indexed, _fts_exhausted = _search_via_index(
+    outcome = _search_via_index(
         base,
         rules,
         query,
         query_lower,
         since_ms=since_ms,
         max_results=max_results,
     )
-    if indexed is None:
-        return _search_live_scan(
+    if outcome.hits is None:
+        return _live_scan_with_index_lock_fallback(
             base,
             rules,
             query,
             query_lower,
             since_ms=since_ms,
             max_results=max_results,
+            index_locked_without_hits=outcome.index_locked_without_hits,
         )
 
-    if len(indexed) >= max_results:
-        return indexed
+    if len(outcome.hits) >= max_results:
+        return outcome.hits
 
-    live = _search_live_scan(
+    live = _live_scan_with_index_lock_fallback(
         base,
         rules,
         query,
         query_lower,
         since_ms=since_ms,
         max_results=max_results,
+        index_locked_without_hits=False,
     )
-    return _merge_search_hits(indexed, live, max_results=max_results)
+    return _merge_search_hits(outcome.hits, live, max_results=max_results)
 
 
 def _search_live_scan(
@@ -359,6 +400,12 @@ def search() -> FlaskReturn:
                 max_results=max_results,
             )
         )
+    except _SearchIndexUnavailableError:
+        return error_response(
+            ErrorCode.SEARCH_INDEX_UNAVAILABLE,
+            "Search index is temporarily unavailable",
+            503,
+        )
     except Exception:
         _logger.exception("Unexpected error during search")
         return error_response(

docs/api-reference.md

@@ -60,6 +60,7 @@ Extra fields may appear for specific codes (for example `since` on invalid bulk-
 | `SEARCH_QUERY_TOO_LONG` | 400 | `GET /api/search` | Query param `q` exceeds 500 characters |
 | `SEARCH_INVALID_SINCE_DAYS` | 400 | `GET /api/search` | Query param `since_days` is not a positive integer |
 | `SEARCH_PROJECTS_UNAVAILABLE` | 503 | `GET /api/search` | Claude projects directory exists but is not readable |
+| `SEARCH_INDEX_UNAVAILABLE` | 503 | `GET /api/search` | FTS index locked during rebuild and live-scan fallback failed |
 | `INVALID_PATH` | 400 | Session, stats, export session | Path traversal or rejected URL segment |
 | `SESSION_NOT_FOUND` | 404 | Session, stats, export session | File missing on disk or session excluded by rules |
 | `INVALID_REQUEST_BODY` | 400 | `POST /api/export` | Body is not a JSON object |
@@ -319,6 +320,7 @@ By default, messages older than 30 days are excluded. Sessions without a parseab
 | 400 | `SEARCH_INVALID_LIMIT` | `limit` not a positive integer (e.g. `abc`, `0`, `1.5`) |
 | 400 | `SEARCH_INVALID_SINCE_DAYS` | `since_days` not a positive integer |
 | 503 | `SEARCH_PROJECTS_UNAVAILABLE` | Projects directory exists but is not readable |
+| 503 | `SEARCH_INDEX_UNAVAILABLE` | FTS index locked during rebuild and live-scan fallback failed |
 | 500 | `INTERNAL_ERROR` | Unexpected server failure |
 
 ```bash

models/error_codes.py

@@ -11,6 +11,7 @@ class ErrorCode(StrEnum):
     SEARCH_QUERY_TOO_LONG = "SEARCH_QUERY_TOO_LONG"
     SEARCH_INVALID_SINCE_DAYS = "SEARCH_INVALID_SINCE_DAYS"
     SEARCH_PROJECTS_UNAVAILABLE = "SEARCH_PROJECTS_UNAVAILABLE"
+    SEARCH_INDEX_UNAVAILABLE = "SEARCH_INDEX_UNAVAILABLE"
     INVALID_PATH = "INVALID_PATH"
     SESSION_NOT_FOUND = "SESSION_NOT_FOUND"
     INVALID_REQUEST_BODY = "INVALID_REQUEST_BODY"

tests/test_error_codes.py

@@ -2,6 +2,9 @@
 
 from __future__ import annotations
 
+import json
+from types import SimpleNamespace
+
 import pytest
 
 from api.error_codes import ErrorCode
@@ -73,3 +76,23 @@ def test_bulk_export_empty_includes_export_nothing_code(client_empty):
     resp = client_empty.post("/api/export", json={"since": "all"})
     assert resp.status_code == 422
     assert_error_response(resp, expected_code="EXPORT_NOTHING_TO_EXPORT")
+
+
+def test_search_index_unavailable_code(client_single, monkeypatch):
+    monkeypatch.setattr(
+        "api.search._search_via_index",
+        lambda *_args, **_kwargs: SimpleNamespace(
+            hits=None,
+            fts_exhausted=False,
+            index_locked_without_hits=True,
+        ),
+    )
+    monkeypatch.setattr(
+        "api.search._search_live_scan",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("live scan failed")),
+    )
+    resp = client_single.get("/api/search?q=test")
+    assert resp.status_code == 503
+    body_text = json.dumps(resp.get_json())
+    assert_error_response(resp, expected_code=ErrorCode.SEARCH_INDEX_UNAVAILABLE)
+    assert "live scan failed" not in body_text

tests/test_error_propagation.py

@@ -33,6 +33,7 @@ class names from a defensive blocklist.
 from flask import Flask
 
 from api.projects import projects_bp
+from api.search import _IndexSearchOutcome
 from api.sessions import sessions_bp
 
 # Defensive blocklist — any of these substrings appearing in a response body
@@ -212,7 +213,10 @@ def test_search_internal_error_does_not_leak(client_single, monkeypatch):
     def _boom(*_args, **_kwargs):
         raise RuntimeError("internal_secret_search_token")
 
-    monkeypatch.setattr("api.search._search_via_index", lambda *_a, **_kw: (None, False))
+    monkeypatch.setattr(
+        "api.search._search_via_index",
+        lambda *_a, **_kw: _IndexSearchOutcome(None, False),
+    )
     monkeypatch.setattr("api.search._search_live_scan", _boom)
 
     resp = client_single.get("/api/search?q=Hello&all_history=1")

tests/test_search.py

@@ -136,6 +136,31 @@ def test_index_lock_falls_back_to_live_scan(tmp_path, monkeypatch):
     assert len(resp.get_json()) >= 1
 
 
+def test_index_lock_returns_unavailable_when_live_scan_fails(tmp_path, monkeypatch):
+    recent_ts = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
+    client = _seed_indexed_client(tmp_path, monkeypatch, timestamp=recent_ts)
+    with (
+        patch(
+            "api.search.query_index_hits",
+            return_value={
+                "hits": [],
+                "query_ok": True,
+                "sql_rows_fetched": 0,
+                "sql_exhausted": True,
+                "index_locked": True,
+            },
+        ),
+        patch(
+            "api.search._search_live_scan",
+            side_effect=RuntimeError("live scan failed"),
+        ),
+    ):
+        resp = client.get("/api/search?q=Hello")
+    assert resp.status_code == 503
+    assert_error_response(resp, expected_code="SEARCH_INDEX_UNAVAILABLE")
+    assert "live scan failed" not in json.dumps(resp.get_json())
+
+
 def _index_patches(cache_root: Path):
     return (patch("utils.search_index.cache_dir", return_value=cache_root),)
 

tests/test_search_index.py

@@ -315,17 +315,18 @@ def test_phrase_filter_paginates_past_decoy_token_matches(self, tmp_path, monkey
         patches = _index_patches(cache_root)
         with patches[0]:
             build_search_index(str(projects), [], force=True)
-            results, _fts_exhausted = _search_via_index(
+            outcome = _search_via_index(
                 str(projects),
                 [],
                 phrase,
                 phrase.lower(),
                 since_ms=None,
                 max_results=1,
             )
-            assert results is not None
-            assert len(results) == 1
-            assert phrase in results[0]["snippet"] or "phrase target" in results[0]["snippet"]
+            assert outcome.hits is not None
+            assert len(outcome.hits) == 1
+            snippet = outcome.hits[0]["snippet"]
+            assert phrase in snippet or "phrase target" in snippet
 
     def test_fts_failure_returns_query_not_ok(self, indexed_tree):
         @contextmanager
@@ -453,17 +454,17 @@ def test_index_search_fills_limit_after_excluded_hits(self, tmp_path, monkeypatc
         patches = _index_patches(cache_root)
         with patches[0]:
             build_search_index(str(projects), rules, force=True)
-            results, _fts_exhausted = _search_via_index(
+            outcome = _search_via_index(
                 str(projects),
                 rules,
                 term,
                 term.lower(),
                 since_ms=None,
                 max_results=50,
             )
-            assert results is not None
-            assert len(results) == 50
-            assert all(hit["project"] == "good-proj" for hit in results)
+            assert outcome.hits is not None
+            assert len(outcome.hits) == 50
+            assert all(hit["project"] == "good-proj" for hit in outcome.hits)
 
     def test_system_content_index_and_live_scan_parity(self, tmp_path, monkeypatch):
         cache_root = tmp_path / "cache"