Upload 8 files

app/bucket_snapshot.py

@@ -21,7 +21,7 @@ from .config import ARCHIVE_END, ARCHIVE_START, DATA_ROOT, ES_URL, INDEX_NAME
 from .data_loader import parsed_archive_revisions, parsed_corpus_fingerprint
 from .doc_store import DOC_DB
 from .facet_store import FACET_DB
-from .indexer import build_fingerprint, desired_metadata
+from .indexer import build_fingerprint, desired_metadata, sidecar_metadata_matches
 from .storage_lock import serving_lock
 
 MANIFEST_NAME = "current.json"
@@ -183,6 +183,8 @@ def _publish(bucket_dir: Path) -> None:
     metadata = client.get(index=INDEX_NAME, id="__meta__").get("_source", {})
     if metadata != desired_metadata(source_fingerprint, revisions):
         raise RuntimeError("cannot snapshot an index with mismatched build or source metadata")
+    if not all(sidecar_metadata_matches(path, metadata) for path in SIDECARS.values()):
+        raise RuntimeError("cannot snapshot mismatched SQLite sidecars")
     physical_name = physical_index(client)
     snapshot = f"bha-{int(time.time())}-{uuid.uuid4().hex[:8]}"
     generation = snapshot
@@ -294,6 +296,9 @@ def validate_restored_sidecars(manifest: dict[str, Any]) -> None:
         raise RuntimeError(f"restored docs count mismatch: expected {expected_docs}, got {docs}")
     if facets <= 0:
         raise RuntimeError(f"restored facets count is invalid: {facets}")
+    metadata = manifest["index_metadata"]
+    if not all(sidecar_metadata_matches(path, metadata) for path in SIDECARS.values()):
+        raise RuntimeError("restored SQLite sidecars have mismatched generation metadata")
 
 
 def restored_sidecars_match(manifest: dict[str, Any]) -> bool:

app/data_loader.py

@@ -30,6 +30,7 @@ def ensure_parsed_data(progress: Callable[[int, int], None] | None = None) -> No
         if (target / ".git").exists():
             run(["git", "remote", "set-url", "origin", repo], target)
             run(["git", "fetch", "--depth", "1", "origin", "parsed"], target)
+            run(["git", "clean", "-fdx"], target)
             run(["git", "checkout", "parsed"], target)
             run(["git", "reset", "--hard", "origin/parsed"], target)
             continue

app/indexer.py

@@ -144,6 +144,15 @@ def write_sidecar_metadata(conn: sqlite3.Connection, metadata: dict) -> None:
         (json.dumps(metadata, ensure_ascii=False, sort_keys=True),),
     )
 
+def write_facet_sidecar_metadata(db_path: Path, metadata: dict) -> None:
+    with sqlite3.connect(db_path) as conn:
+        conn.execute("CREATE TABLE IF NOT EXISTS build_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
+        conn.execute(
+            "INSERT OR REPLACE INTO build_meta(key, value) VALUES ('index_metadata', ?)",
+            (json.dumps(metadata, ensure_ascii=False, sort_keys=True),),
+        )
+        conn.commit()
+
 def sidecar_metadata_matches(db_path: Path, metadata: dict) -> bool:
     try:
         with sqlite3.connect(db_path) as conn:
@@ -209,7 +218,7 @@ def incremental_build(es: Elasticsearch, source_fingerprint: str, revisions: dic
     try:
         copy_sqlite(DOC_DB, temp_doc_db)
         copy_sqlite(FACET_DB, temp_facet_db)
-        if not sidecar_metadata_matches(DOC_DB, old_meta):
+        if not sidecar_metadata_matches(DOC_DB, old_meta) or not sidecar_metadata_matches(FACET_DB, old_meta):
             raise IncrementalUnsafe("serving sidecar generation is unavailable")
         old_docs: dict[str, dict] = {}
         new_docs: dict[str, dict] = {}
@@ -245,6 +254,7 @@ def incremental_build(es: Elasticsearch, source_fingerprint: str, revisions: dic
         with connect_doc_db(temp_doc_db) as conn:
             write_sidecar_metadata(conn, desired_metadata(source_fingerprint, revisions))
             conn.commit()
+        write_facet_sidecar_metadata(temp_facet_db, desired_metadata(source_fingerprint, revisions))
         es.indices.refresh(index=build_index)
         with connect_doc_db(temp_doc_db) as conn:
             sqlite_count = int(conn.execute("SELECT COUNT(*) FROM docs").fetchone()[0])
@@ -270,7 +280,7 @@ def incremental_build(es: Elasticsearch, source_fingerprint: str, revisions: dic
             reset_db(temp_doc_db)
             reset_facet_db(temp_facet_db)
 
-def sidecars_ready() -> bool:
+def sidecars_ready(metadata: dict | None = None) -> bool:
     if not DOC_DB.exists() or not FACET_DB.exists():
         return False
     try:
@@ -280,13 +290,16 @@ def sidecars_ready() -> bool:
             meta_count = int(conn.execute("SELECT COUNT(*) FROM build_meta").fetchone()[0])
         with sqlite3.connect(FACET_DB) as conn:
             facet_count = int(conn.execute("SELECT COUNT(*) FROM facets").fetchone()[0])
-        return doc_count > 0 and facet_count > 0 and meta_count > 0
+        docs_match = metadata is None or sidecar_metadata_matches(DOC_DB, metadata)
+        facets_match = metadata is None or sidecar_metadata_matches(FACET_DB, metadata)
+        return doc_count > 0 and facet_count > 0 and meta_count > 0 and docs_match and facets_match
     except Exception:
         return False
 
 def serving_data_available(es: Elasticsearch) -> bool:
     try:
-        return bool(es.indices.exists(index=INDEX_NAME)) and int(es.count(index=INDEX_NAME).get("count", 0)) > 0 and sidecars_ready()
+        metadata = es.get(index=INDEX_NAME, id="__meta__").get("_source", {})
+        return bool(es.indices.exists(index=INDEX_NAME)) and int(es.count(index=INDEX_NAME).get("count", 0)) > 0 and sidecars_ready(metadata)
     except Exception:
         return False
 
@@ -475,7 +488,7 @@ def _ensure_index(reset: bool = False) -> None:
     has_serving_data = serving_data_available(es)
     if es.indices.exists(index=INDEX_NAME):
         count = es.count(index=INDEX_NAME).get("count", 0)
-        if not force_rebuild and count and metadata_matches(es, source_fingerprint, archive_revisions) and sidecars_ready():
+        if not force_rebuild and count and metadata_matches(es, source_fingerprint, archive_revisions) and sidecars_ready(desired_metadata(source_fingerprint, archive_revisions)):
             write_progress(status="ready", indexed=count, errors=0, elapsed_seconds=int(time.time() - started_at))
             return
     if not force_rebuild and has_serving_data:
@@ -576,6 +589,7 @@ def _ensure_index(reset: bool = False) -> None:
         with connect_doc_db(temp_doc_db) as conn:
             write_sidecar_metadata(conn, desired_metadata(source_fingerprint, archive_revisions))
             conn.commit()
+        write_facet_sidecar_metadata(temp_facet_db, desired_metadata(source_fingerprint, archive_revisions))
         es.indices.refresh(index=build_index)
         try:
             es.indices.forcemerge(index=build_index, max_num_segments=1, request_timeout=3600)

app/main.py

@@ -13,7 +13,7 @@ import tempfile
 import threading
 import time
 from urllib.parse import quote, unquote_to_bytes, urlparse
-from typing import Any
+from typing import Any, Literal
 from zipfile import ZIP_DEFLATED, ZipFile
 import httpx
 from elasticsearch import Elasticsearch
@@ -54,6 +54,10 @@ try:
     )
 except ValueError:
     PROOFREAD_PROXY_NETWORKS = ()
+try:
+    PROOFREAD_PROXY_HOPS = max(1, int(os.environ.get("BHA_TRUST_PROXY_HOPS", "1")))
+except ValueError:
+    PROOFREAD_PROXY_HOPS = 1
 INDEX_STATUS_PATH = DATA_ROOT / "index-status"
 INDEX_SWITCH_STATE_PATH = DATA_ROOT / "index-switch.json"
 BUCKET_DIR = Path(os.environ["BHA_BUCKET_DIR"]) if os.environ.get("BHA_BUCKET_DIR") else None
@@ -74,6 +78,7 @@ app.add_middleware(
 es = Elasticsearch(ES_URL, request_timeout=120)
 source_client = httpx.AsyncClient(
     follow_redirects=False,
+    trust_env=False,
     timeout=httpx.Timeout(120, connect=10, pool=10),
     limits=httpx.Limits(max_connections=20, max_keepalive_connections=10, keepalive_expiry=600),
 )
@@ -150,20 +155,20 @@ class SearchRequest(BaseModel):
     exact: bool = True
     fulltext: bool = True
     source: str | None = None
-    sources: list[str] = Field(default_factory=list)
-    exclude_sources: list[str] = Field(default_factory=list)
+    sources: list[str] = Field(default_factory=list, max_length=100)
+    exclude_sources: list[str] = Field(default_factory=list, max_length=100)
     author: str | None = None
-    authors: list[str] = Field(default_factory=list)
-    exclude_authors: list[str] = Field(default_factory=list)
+    authors: list[str] = Field(default_factory=list, max_length=100)
+    exclude_authors: list[str] = Field(default_factory=list, max_length=100)
     tag: str | None = None
-    tags: list[str] = Field(default_factory=list)
-    exclude_tags: list[str] = Field(default_factory=list)
+    tags: list[str] = Field(default_factory=list, max_length=100)
+    exclude_tags: list[str] = Field(default_factory=list, max_length=100)
     archive_id: int | None = None
     publication_type: str | None = None
     date_from: int | None = None
     date_to: int | None = None
     date_terms: list[dict[str, int | str | None]] = Field(default_factory=list)
-    sort: str = "relevance"
+    sort: Literal["relevance", "date_desc", "date_asc", "source", "title"] = "relevance"
 
 class ProofreadRequest(BaseModel):
     doc_id: str = Field(min_length=1, max_length=512)
@@ -618,11 +623,16 @@ def index_status() -> str:
 def cached_status_payload(include_es: bool = False) -> dict[str, Any]:
     global status_response_cache
     now = time.monotonic()
-    if status_response_cache and now - status_response_cache[0] < STATUS_CACHE_TTL_SECONDS:
+    current_status = index_status()
+    if (
+        status_response_cache
+        and now - status_response_cache[0] < STATUS_CACHE_TTL_SECONDS
+        and status_response_cache[1].get("index_status") == current_status
+    ):
         result = dict(status_response_cache[1])
     else:
-        status = index_status()
-        ready = index_ready() and FACET_DB.exists()
+        status = current_status
+        ready = status == "ready" and index_ready() and FACET_DB.exists()
         result = {
             "ok": ready,
             "index_ready": ready,
@@ -725,6 +735,8 @@ def sources():
 
 def search(body: SearchRequest):
     status = index_status()
+    if status == "unknown":
+        status = str(cached_status_payload().get("index_status") or status)
     has_data = status not in {"switching", "restoring", "failed"} and has_index_data()
     if not has_data:
         return JSONResponse({"results": [], "total": 0, "page": max(1, body.page), "page_size": body.page_size, "indexing": status != "ready", "index_status": status})
@@ -733,7 +745,7 @@ def search(body: SearchRequest):
     cache_key = search_cache_key(body, page, page_size)
     cached = search_response_cache.get(cache_key)
     now = time.monotonic()
-    if cached and now - cached[0] < SEARCH_CACHE_TTL_SECONDS:
+    if status == "ready" and cached and now - cached[0] < SEARCH_CACHE_TTL_SECONDS:
         return JSONResponse(dict(cached[1]))
     if (page - 1) * page_size >= MAX_RESULT_WINDOW:
         return JSONResponse({"error": "result window exceeded", "max_result_window": MAX_RESULT_WINDOW}, status_code=400)
@@ -776,7 +788,8 @@ def search(body: SearchRequest):
     except SearchStoreError:
         return JSONResponse({"error": "search storage unavailable"}, status_code=503)
     result = {"results": [trim_hit(hit, stored_docs.get(hit.get("_id")), body.q, body.fulltext, body.exact) for hit in hits if stored_docs.get(hit.get("_id"))], "total": total_value, "page": page, "page_size": page_size, "indexing": status != "ready", "index_status": status}
-    search_response_cache[cache_key] = (time.monotonic(), result)
+    if status == "ready":
+        search_response_cache[cache_key] = (time.monotonic(), result)
     if len(search_response_cache) > SEARCH_CACHE_MAX_ENTRIES:
         oldest_key = min(search_response_cache, key=lambda item: search_response_cache[item][0])
         search_response_cache.pop(oldest_key, None)
@@ -849,9 +862,16 @@ def proofread_rate_error(request: Request) -> JSONResponse | None:
     except ValueError:
         pass
     forwarded = request.headers.get("x-forwarded-for", "") if trusted_peer else ""
-    real_ip = request.headers.get("x-real-ip", "") if trusted_peer else ""
     candidates = [value.strip() for value in forwarded.split(",") if value.strip()]
-    key = real_ip.strip() if real_ip.strip() else (candidates[0] if candidates else peer)
+    forwarded_client = ""
+    if candidates and len(candidates) > PROOFREAD_PROXY_HOPS:
+        candidate = candidates[-PROOFREAD_PROXY_HOPS - 1]
+        try:
+            if ipaddress.ip_address(candidate):
+                forwarded_client = candidate
+        except ValueError:
+            pass
+    key = forwarded_client or peer
     now = time.monotonic()
     with proofread_rate_lock:
         recent = [value for value in proofread_requests.get(key, []) if now - value < PROOFREAD_RATE_WINDOW_SECONDS]
@@ -867,8 +887,33 @@ def proofread_rate_error(request: Request) -> JSONResponse | None:
                     expired.append(identity)
             for identity in expired:
                 proofread_requests.pop(identity, None)
+            while len(proofread_requests) > 10000:
+                oldest = min(proofread_requests, key=lambda identity: min(proofread_requests[identity]))
+                proofread_requests.pop(oldest, None)
+    try:
+        request.state.proofread_rate_key = key
+        request.state.proofread_rate_timestamp = now
+    except Exception:
+        pass
     return None
 
+def release_proofread_rate(request: Request) -> None:
+    try:
+        key = request.state.proofread_rate_key
+        timestamp = request.state.proofread_rate_timestamp
+    except Exception:
+        return
+    with proofread_rate_lock:
+        values = proofread_requests.get(key, [])
+        try:
+            values.remove(timestamp)
+        except ValueError:
+            return
+        if values:
+            proofread_requests[key] = values
+        else:
+            proofread_requests.pop(key, None)
+
 def clean_string_list(value: Any, field: str, limit: int) -> list[str]:
     if not isinstance(value, list) or len(value) > limit:
         raise ValueError(f"{field} must be a list with at most {limit} entries")
@@ -1187,6 +1232,7 @@ async def submit_proofread(body: ProofreadRequest, request: Request):
     try:
         await dispatch_proofread(payload)
     except Exception as exc:
+        release_proofread_rate(request)
         return JSONResponse({"error": str(exc)}, status_code=502)
     return {"ok": True, "status": "accepted", "tasks": 1, "auto_merge": payload["auto_merge"]}
 
@@ -1474,6 +1520,8 @@ def reindex_status(request: Request):
         pass
     return {
         "index_status": index_status(),
+        "progress": index_progress(),
+        "switch_pending": INDEX_SWITCH_STATE_PATH.exists(),
         "archive_revisions": metadata.get("archive_revisions") or {},
         "build_fingerprint": metadata.get("build_fingerprint"),
     }

app/search_store.py

@@ -58,16 +58,16 @@ def get_search_docs(doc_ids: list[str], query: str, fulltext: bool, exact: bool)
                     FROM requested r
                     JOIN docs d ON d.doc_id = r.doc_id
                 )
-                SELECT doc_id, publication_name, publication_type, title, authors_json,
-                       tag_names_json, date_display_json, source_file_count,
-                       substr(
-                           content,
+                 SELECT doc_id, publication_name, publication_type, title, authors_json,
+                        tag_names_json, date_display_json, source_file_count,
+                        substr(
+                            content,
                            CASE WHEN match_pos IS NULL THEN 1 ELSE max(1, match_pos - ?) END,
                            CASE WHEN match_pos IS NULL THEN ? ELSE ? END
                        ) AS content,
                        CASE WHEN match_pos IS NULL THEN 1 ELSE max(1, match_pos - ?) END AS content_window_start,
-                       content_length
-                FROM located
+                        content_length
+                 FROM located
                 ORDER BY result_order
                 """,
                 [

app/storage_lock.py

@@ -1,13 +1,11 @@
 import fcntl
 import errno
-import threading
 from contextlib import contextmanager
 from typing import Iterator
 
 from .config import DATA_ROOT
 
 LOCK_PATH = DATA_ROOT / "serving.lock"
-PROCESS_LOCK = threading.RLock()
 
 @contextmanager
 def serving_lock(exclusive: bool = False) -> Iterator[None]:
@@ -20,9 +18,7 @@ def serving_lock(exclusive: bool = False) -> Iterator[None]:
         lock_file.close()
         if exc.errno not in {errno.ENOSYS, errno.EOPNOTSUPP, errno.ENOTSUP}:
             raise
-        with PROCESS_LOCK:
-            yield
-        return
+        raise RuntimeError("serving lock requires filesystem flock support") from exc
     try:
         yield
     finally: