Upload 5 files

app/bucket_probe.py

@@ -3,14 +3,24 @@
 from __future__ import annotations
 
 import sys
+import shutil
 import uuid
 from pathlib import Path
 
 from elasticsearch import Elasticsearch
 
-from .config import ES_URL
+from .config import DATA_ROOT, ES_URL
 
-SNAPSHOT_ROOT = "snapshots/probe"
+LOCAL_REPOSITORY = DATA_ROOT / "es-repositories" / "probe"
+
+
+def clear_directory(path: Path) -> None:
+    path.mkdir(parents=True, exist_ok=True)
+    for child in path.iterdir():
+        if child.is_dir():
+            shutil.rmtree(child)
+        else:
+            child.unlink()
 
 
 def main() -> int:
@@ -22,17 +32,27 @@ def main() -> int:
     repository = "bha_bucket_probe"
     index = f"bha-bucket-probe-{uuid.uuid4().hex[:12]}"
     snapshot = f"probe-{uuid.uuid4().hex[:12]}"
+    bucket_copy = bucket_dir / "probe" / snapshot
     client = Elasticsearch(ES_URL, request_timeout=120)
     try:
+        clear_directory(LOCAL_REPOSITORY)
         client.indices.create(index=index)
         client.index(index=index, id="1", document={"probe": "bucket"}, refresh="wait_for")
         client.snapshot.create_repository(
             name=repository,
-            repository={"type": "fs", "settings": {"location": str(bucket_dir / SNAPSHOT_ROOT)}},
+            repository={"type": "fs", "settings": {"location": str(LOCAL_REPOSITORY)}},
             verify=True,
         )
         client.snapshot.create(repository=repository, snapshot=snapshot, indices=index, wait_for_completion=True)
+        shutil.copytree(LOCAL_REPOSITORY, bucket_copy)
         client.indices.delete(index=index)
+        clear_directory(LOCAL_REPOSITORY)
+        shutil.copytree(bucket_copy, LOCAL_REPOSITORY, dirs_exist_ok=True)
+        client.snapshot.create_repository(
+            name=repository,
+            repository={"type": "fs", "settings": {"location": str(LOCAL_REPOSITORY)}},
+            verify=True,
+        )
         client.snapshot.restore(repository=repository, snapshot=snapshot, indices=index, wait_for_completion=True)
         if int(client.count(index=index)["count"]) != 1:
             raise RuntimeError("restored probe index has the wrong document count")
@@ -40,6 +60,8 @@ def main() -> int:
         return 0
     finally:
         client.indices.delete(index=index, ignore_unavailable=True)
+        shutil.rmtree(bucket_copy, ignore_errors=True)
+        clear_directory(LOCAL_REPOSITORY)
 
 
 if __name__ == "__main__":

app/bucket_snapshot.py

@@ -8,7 +8,6 @@ import os
 import re
 import shutil
 import sqlite3
-import subprocess
 import sys
 import tempfile
 import time
@@ -18,39 +17,32 @@ from typing import Any
 
 from elasticsearch import Elasticsearch
 
-from .config import ARCHIVE_END, ARCHIVE_START, DATA_ROOT, ES_URL, INDEX_NAME, INDEX_VERSION, REPO_PREFIX
+from .config import DATA_ROOT, ES_URL, INDEX_NAME
+from .data_loader import parsed_corpus_fingerprint
 from .doc_store import DOC_DB
 from .facet_store import FACET_DB
+from .indexer import build_fingerprint, desired_metadata
 from .storage_lock import serving_lock
 
 MANIFEST_NAME = "current.json"
 REPOSITORY = "bha_bucket"
-# Keep production repository metadata separate from the disposable probe repository.
-SNAPSHOT_ROOT = "snapshots/production"
 GENERATIONS_ROOT = "generations"
+REPOSITORY_DIRNAME = "elasticsearch"
 SIDECARS = {"docs": DOC_DB, "facets": FACET_DB}
 SNAPSHOT_NAME_RE = re.compile(r"^bha-[a-z0-9-]+$")
 
 
-def expected_metadata() -> dict[str, Any]:
-    return {"version": INDEX_VERSION, "archive_start": ARCHIVE_START, "archive_end": ARCHIVE_END}
+def local_repository() -> Path:
+    return DATA_ROOT / "es-repositories" / "production"
 
 
-def corpus_fingerprint(remote: bool) -> str:
-    revisions: list[str] = []
-    for archive_id in range(ARCHIVE_START, ARCHIVE_END + 1):
-        if remote:
-            command = ["git", "ls-remote", f"{REPO_PREFIX}/banned-historical-archives{archive_id}.git", "refs/heads/parsed"]
-            result = subprocess.run(command, check=True, capture_output=True, text=True)
-            revision = result.stdout.split(maxsplit=1)[0]
+def clear_directory(path: Path) -> None:
+    path.mkdir(parents=True, exist_ok=True)
+    for child in path.iterdir():
+        if child.is_dir():
+            shutil.rmtree(child)
         else:
-            repository_path = DATA_ROOT / "parsed" / f"archives{archive_id}"
-            result = subprocess.run(["git", "rev-parse", "HEAD"], cwd=repository_path, check=True, capture_output=True, text=True)
-            revision = result.stdout.strip()
-        if not re.fullmatch(r"[0-9a-f]{40}", revision):
-            raise RuntimeError(f"invalid parsed revision for archive {archive_id}")
-        revisions.append(f"{archive_id}:{revision}")
-    return hashlib.sha256("\n".join(revisions).encode()).hexdigest()
+            child.unlink()
 
 
 def sha256(path: Path) -> str:
@@ -65,10 +57,10 @@ def file_metadata(path: Path) -> dict[str, Any]:
     return {"filename": path.name, "bytes": path.stat().st_size, "sha256": sha256(path)}
 
 
-def repository(client: Elasticsearch, bucket_dir: Path) -> None:
+def repository(client: Elasticsearch) -> None:
     client.snapshot.create_repository(
         name=REPOSITORY,
-        repository={"type": "fs", "settings": {"location": str(bucket_dir / SNAPSHOT_ROOT)}},
+        repository={"type": "fs", "settings": {"location": str(local_repository())}},
         verify=True,
     )
 
@@ -82,12 +74,15 @@ def read_manifest(bucket_dir: Path) -> dict[str, Any] | None:
         data = json.loads(manifest_path(bucket_dir).read_text(encoding="utf-8"))
     except (OSError, json.JSONDecodeError):
         return None
-    if not isinstance(data, dict) or data.get("schema") != 1 or data.get("index_metadata") != expected_metadata():
+    if not isinstance(data, dict) or data.get("schema") != 2:
+        return None
+    index_metadata = data.get("index_metadata")
+    if not isinstance(index_metadata, dict) or index_metadata.get("build_fingerprint") != build_fingerprint():
         return None
     snapshot = data.get("snapshot")
     generation = data.get("generation")
     physical_index = data.get("physical_index")
-    source_fingerprint = data.get("source_fingerprint")
+    source_fingerprint = index_metadata.get("source_fingerprint")
     sidecars = data.get("sidecars")
     if not isinstance(snapshot, str) or not SNAPSHOT_NAME_RE.fullmatch(snapshot):
         return None
@@ -97,6 +92,8 @@ def read_manifest(bucket_dir: Path) -> dict[str, Any] | None:
         return None
     if not isinstance(source_fingerprint, str) or not re.fullmatch(r"[0-9a-f]{64}", source_fingerprint):
         return None
+    if index_metadata != desired_metadata(source_fingerprint):
+        return None
     if not isinstance(sidecars, dict) or set(sidecars) != set(SIDECARS):
         return None
     for key, target in SIDECARS.items():
@@ -125,14 +122,15 @@ def publish(bucket_dir: Path) -> None:
     if not all(path.is_file() for path in SIDECARS.values()):
         raise RuntimeError("cannot snapshot missing SQLite sidecars")
     client = Elasticsearch(ES_URL, request_timeout=3600)
+    source_fingerprint = parsed_corpus_fingerprint()
     metadata = client.get(index=INDEX_NAME, id="__meta__").get("_source", {})
-    if metadata != expected_metadata():
-        raise RuntimeError("cannot snapshot an index with mismatched metadata")
-    source_fingerprint = corpus_fingerprint(remote=False)
+    if metadata != desired_metadata(source_fingerprint):
+        raise RuntimeError("cannot snapshot an index with mismatched build or source metadata")
     physical_name = physical_index(client)
     snapshot = f"bha-{int(time.time())}-{uuid.uuid4().hex[:8]}"
     generation = snapshot
-    repository(client, bucket_dir)
+    clear_directory(local_repository())
+    repository(client)
     started = time.monotonic()
     client.snapshot.create(
         repository=REPOSITORY,
@@ -143,6 +141,7 @@ def publish(bucket_dir: Path) -> None:
     )
     generation_dir = bucket_dir / GENERATIONS_ROOT / generation
     generation_dir.mkdir(parents=True, exist_ok=False)
+    shutil.copytree(local_repository(), generation_dir / REPOSITORY_DIRNAME)
     sidecars: dict[str, dict[str, Any]] = {}
     with tempfile.TemporaryDirectory(dir=DATA_ROOT, prefix="bha-snapshot-") as temporary_dir:
         for key, source in SIDECARS.items():
@@ -152,11 +151,9 @@ def publish(bucket_dir: Path) -> None:
             shutil.copyfile(local_copy, destination)
             sidecars[key] = file_metadata(destination)
     payload = {
-        "schema": 1,
+        "schema": 2,
         "index_name": INDEX_NAME,
-        "index_version": INDEX_VERSION,
-        "index_metadata": expected_metadata(),
-        "source_fingerprint": source_fingerprint,
+        "index_metadata": metadata,
         "physical_index": physical_name,
         "snapshot": snapshot,
         "generation": generation,
@@ -171,13 +168,12 @@ def publish(bucket_dir: Path) -> None:
         for path in (bucket_dir / GENERATIONS_ROOT).iterdir():
             if path.is_dir() and path.name != generation and SNAPSHOT_NAME_RE.fullmatch(path.name):
                 shutil.rmtree(path)
-        snapshots = client.snapshot.get(repository=REPOSITORY, snapshot="_all").get("snapshots", [])
-        for previous in snapshots:
-            previous_name = previous.get("snapshot")
-            if isinstance(previous_name, str) and previous_name != snapshot:
-                client.snapshot.delete(repository=REPOSITORY, snapshot=previous_name)
+        legacy_repository = bucket_dir / "snapshots"
+        if legacy_repository.exists():
+            shutil.rmtree(legacy_repository)
     except Exception as exc:
         print(f"bha_bucket_snapshot=cleanup_failed error={type(exc).__name__}", flush=True)
+    clear_directory(local_repository())
     print(f"bha_bucket_snapshot=published snapshot={snapshot} elapsed_seconds={time.monotonic() - started:.2f}", flush=True)
 
 
@@ -238,19 +234,23 @@ def restore(bucket_dir: Path) -> bool:
         print("bha_bucket_snapshot=unavailable reason=missing_or_invalid_manifest", flush=True)
         return False
     try:
-        if manifest["source_fingerprint"] != corpus_fingerprint(remote=True):
+        if manifest["index_metadata"]["source_fingerprint"] != parsed_corpus_fingerprint(remote=True):
             print("bha_bucket_snapshot=unavailable reason=source_fingerprint_mismatch", flush=True)
             return False
     except Exception as exc:
-        print(f"bha_bucket_snapshot=unavailable reason=source_fingerprint_{type(exc).__name__}", flush=True)
-        return False
+        print(f"bha_bucket_snapshot=source_fingerprint_unverified reason={type(exc).__name__}", flush=True)
     client = Elasticsearch(ES_URL, request_timeout=3600)
     if client.indices.exists(index=INDEX_NAME):
         print("bha_bucket_snapshot=skipped reason=index_exists", flush=True)
         return False
     started = time.monotonic()
     try:
-        repository(client, bucket_dir)
+        clear_directory(local_repository())
+        repository_source = bucket_dir / GENERATIONS_ROOT / str(manifest["generation"]) / REPOSITORY_DIRNAME
+        if not repository_source.is_dir():
+            raise RuntimeError("bucket snapshot repository is missing")
+        shutil.copytree(repository_source, local_repository(), dirs_exist_ok=True)
+        repository(client)
         client.snapshot.restore(
             repository=REPOSITORY,
             snapshot=str(manifest["snapshot"]),
@@ -259,17 +259,25 @@ def restore(bucket_dir: Path) -> bool:
             include_aliases=True,
             wait_for_completion=True,
         )
+        if not client.indices.exists_alias(name=INDEX_NAME):
+            client.indices.put_alias(index=str(manifest["physical_index"]), name=INDEX_NAME)
         metadata = client.get(index=INDEX_NAME, id="__meta__").get("_source", {})
         count = int(client.count(index=INDEX_NAME, query={"exists": {"field": "doc_id"}})["count"])
-        if metadata != expected_metadata() or count != int(manifest["document_count"]):
+        if metadata != manifest["index_metadata"] or count != int(manifest["document_count"]):
             raise RuntimeError("restored Elasticsearch snapshot failed validation")
         restore_sidecars(bucket_dir, manifest)
         if not restored_sidecars_match(manifest):
             raise RuntimeError("restored SQLite sidecars failed validation")
     except Exception as exc:
-        client.indices.delete(index=INDEX_NAME, ignore_unavailable=True)
+        for target in (INDEX_NAME, str(manifest["physical_index"])):
+            try:
+                client.indices.delete(index=target, ignore_unavailable=True)
+            except Exception:
+                pass
+        clear_directory(local_repository())
         print(f"bha_bucket_snapshot=unavailable reason={type(exc).__name__}", flush=True)
         return False
+    clear_directory(local_repository())
     print(f"bha_bucket_snapshot=restored snapshot={manifest['snapshot']} elapsed_seconds={time.monotonic() - started:.2f}", flush=True)
     return True
 

app/data_loader.py

@@ -1,3 +1,4 @@
+import hashlib
 import json
 import re
 import shutil
@@ -35,6 +36,31 @@ def ensure_parsed_data(progress: Callable[[int, int], None] | None = None) -> No
             shutil.rmtree(target)
         run(["git", "clone", "--depth", "1", "--branch", "parsed", repo, str(target)])
 
+def parsed_corpus_fingerprint(remote: bool = False) -> str:
+    revisions: list[str] = []
+    for archive_id in range(ARCHIVE_START, ARCHIVE_END + 1):
+        if remote:
+            result = subprocess.run(
+                ["git", "ls-remote", f"{REPO_PREFIX}/banned-historical-archives{archive_id}.git", "refs/heads/parsed"],
+                check=True,
+                capture_output=True,
+                text=True,
+            )
+            revision = result.stdout.split(maxsplit=1)[0]
+        else:
+            result = subprocess.run(
+                ["git", "rev-parse", "HEAD"],
+                cwd=PARSED_ROOT / f"archives{archive_id}",
+                check=True,
+                capture_output=True,
+                text=True,
+            )
+            revision = result.stdout.strip()
+        if not re.fullmatch(r"[0-9a-f]{40}", revision):
+            raise RuntimeError(f"invalid parsed revision for archive {archive_id}")
+        revisions.append(f"{archive_id}:{revision}")
+    return hashlib.sha256("\n".join(revisions).encode()).hexdigest()
+
 def read_json(path: Path) -> Any:
     try:
         return json.loads(path.read_text(encoding="utf-8"))

app/indexer.py

@@ -1,5 +1,6 @@
 import argparse
 import fcntl
+import hashlib
 import json
 import os
 import sqlite3
@@ -11,7 +12,7 @@ from typing import IO
 from elasticsearch import Elasticsearch, helpers
 from elasticsearch.exceptions import NotFoundError
 from .config import DATA_ROOT, ARCHIVE_END, ARCHIVE_START, ES_URL, INDEX_NAME, INDEX_VERSION, RESET_INDEX
-from .data_loader import ensure_parsed_data, iter_documents
+from .data_loader import ensure_parsed_data, iter_documents, parsed_corpus_fingerprint
 from .doc_store import DOC_DB, connect as connect_doc_db, init_db as init_doc_db, reset_db, write_docs_to_conn
 from .facet_store import FACET_DB, connect as connect_facet_db, init_db as init_facet_db, reset_db as reset_facet_db, write_facets
 from .storage_lock import serving_lock
@@ -93,16 +94,36 @@ def index_mapping() -> dict:
         },
     }
 
-def desired_metadata() -> dict:
-    return {"version": INDEX_VERSION, "archive_start": ARCHIVE_START, "archive_end": ARCHIVE_END}
+def build_fingerprint() -> str:
+    digest = hashlib.sha256(json.dumps(index_mapping(), sort_keys=True, separators=(",", ":")).encode())
+    for module_path in (
+        Path(__file__),
+        Path(__file__).with_name("data_loader.py"),
+        Path(__file__).with_name("doc_store.py"),
+        Path(__file__).with_name("facet_store.py"),
+    ):
+        digest.update(module_path.name.encode())
+        digest.update(module_path.read_bytes())
+    return digest.hexdigest()
 
-def metadata_matches(es: Elasticsearch) -> bool:
+def desired_metadata(source_fingerprint: str | None = None) -> dict:
+    metadata = {
+        "version": INDEX_VERSION,
+        "archive_start": ARCHIVE_START,
+        "archive_end": ARCHIVE_END,
+        "build_fingerprint": build_fingerprint(),
+    }
+    if source_fingerprint is not None:
+        metadata["source_fingerprint"] = source_fingerprint
+    return metadata
+
+def metadata_matches(es: Elasticsearch, source_fingerprint: str) -> bool:
     try:
         data = es.get(index=INDEX_NAME, id="__meta__")
     except Exception:
         return False
     source = data.get("_source", {})
-    return all(source.get(key) == value for key, value in desired_metadata().items())
+    return source == desired_metadata(source_fingerprint)
 
 def sidecars_ready() -> bool:
     if not DOC_DB.exists() or not FACET_DB.exists():
@@ -282,13 +303,14 @@ def _ensure_index(reset: bool = False) -> None:
         archive_total=total,
         elapsed_seconds=int(time.time() - started_at),
     ))
+    source_fingerprint = parsed_corpus_fingerprint()
     parsed_elapsed_seconds = round(time.time() - parsed_started_at, 2)
     print(f"index_phase=parsed_data elapsed_seconds={parsed_elapsed_seconds}")
     force_rebuild = reset or RESET_INDEX
     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) and sidecars_ready():
+        if not force_rebuild and count and metadata_matches(es, source_fingerprint) and sidecars_ready():
             print(f"{INDEX_NAME} already has {count} documents")
             write_progress(status="ready", indexed=count, errors=0, elapsed_seconds=int(time.time() - started_at))
             return
@@ -370,7 +392,7 @@ def _ensure_index(reset: bool = False) -> None:
         write_facets(facet_counters, temp_facet_db)
         facets_elapsed_seconds = round(time.time() - facets_started_at, 2)
         print(f"index_phase=facets elapsed_seconds={facets_elapsed_seconds}")
-        es.index(index=build_index, id="__meta__", document=desired_metadata())
+        es.index(index=build_index, id="__meta__", document=desired_metadata(source_fingerprint))
         refresh_started_at = time.time()
         es.indices.refresh(index=build_index)
         refresh_elapsed_seconds = round(time.time() - refresh_started_at, 2)

app/main.py

@@ -1,5 +1,6 @@
 from pathlib import Path
 from html import escape
+import hmac
 import ipaddress
 import asyncio
 import json
@@ -31,8 +32,9 @@ from .indexer import acquire_index_lock, ensure_index
 from .bucket_snapshot import publish as publish_bucket_snapshot
 from .storage_lock import serving_lock
 MAX_RESULT_WINDOW = 500000
-REINDEX_COOLDOWN_SECONDS = 30 * 24 * 60 * 60
+REINDEX_COOLDOWN_SECONDS = 10 * 60
 REINDEX_LAST_START = DATA_ROOT / "reindex-last-start"
+REINDEX_TOKEN = os.environ.get("BHA_REINDEX_TOKEN", "")
 INDEX_STATUS_PATH = DATA_ROOT / "index-status"
 BUCKET_DIR = Path(os.environ["BHA_BUCKET_DIR"]) if os.environ.get("BHA_BUCKET_DIR") else None
 app = FastAPI(title="BHA Search Lite")
@@ -655,7 +657,7 @@ def facet(kind: str, request: Request, page: int = 1, page_size: int = 200, q: s
     status_data = cached_status_payload()
     timings["status"] = (time.perf_counter() - status_started) * 1000
     status = str(status_data.get("index_status") or "unknown")
-    if not FACET_DB.exists():
+    if status == "restoring" or not FACET_DB.exists():
         return {"items": [], "total": 0, "page": max(1, page), "page_size": page_size, "has_more": False, "indexing": status != "ready", "index_status": status}
     lock_started = time.perf_counter()
     with serving_lock():
@@ -671,7 +673,7 @@ def sources(request: Request):
     status_data = cached_status_payload()
     timings["status"] = (time.perf_counter() - status_started) * 1000
     status = str(status_data.get("index_status") or "unknown")
-    if not FACET_DB.exists():
+    if status == "restoring" or not FACET_DB.exists():
         return {"sources": [], "authors": [], "tags": [], "archives": [], "types": [], "years": [], "indexing": status != "ready", "index_status": status}
     lock_started = time.perf_counter()
     with serving_lock():
@@ -704,7 +706,7 @@ def search(body: SearchRequest, request: Request):
     timings = request.state.server_timings
     status_started = time.perf_counter()
     status = index_status()
-    has_data = status != "switching" and has_index_data()
+    has_data = status not in {"switching", "restoring"} and has_index_data()
     timings["status"] = (time.perf_counter() - status_started) * 1000
     if not has_data:
         return timed_json_response(
@@ -783,6 +785,8 @@ def sort_clause(sort_key: str) -> list[dict[str, Any]] | None:
 
 def preview(doc_id: str, request: Request):
     timings = request.state.server_timings
+    if index_status() == "restoring":
+        return JSONResponse({"error": "index restore in progress"}, status_code=503)
     lock_started = time.perf_counter()
     with serving_lock():
         timings["lock"] = (time.perf_counter() - lock_started) * 1000
@@ -799,6 +803,8 @@ def preview(doc_id: str, request: Request):
 
 def random_preview(request: Request):
     timings = request.state.server_timings
+    if index_status() == "restoring":
+        return JSONResponse({"error": "index restore in progress"}, status_code=503)
     lock_started = time.perf_counter()
     with serving_lock():
         timings["lock"] = (time.perf_counter() - lock_started) * 1000
@@ -1036,9 +1042,26 @@ def run_reindex(lock_file) -> None:
     finally:
         clear_response_caches()
         lock_file.close()
+
+def start_reindex_worker(lock_file) -> None:
+    threading.Thread(target=run_reindex, args=(lock_file,), daemon=True).start()
+
 @app.post("/api/reindex", status_code=202)
 
-def reindex():
+def reindex(request: Request):
+    if not REINDEX_TOKEN:
+        return JSONResponse(
+            {"error": "manual reindex is not configured"},
+            status_code=503,
+            headers={"Cache-Control": "no-store"},
+        )
+    scheme, _, supplied_token = request.headers.get("Authorization", "").partition(" ")
+    if scheme.lower() != "bearer" or not supplied_token or not hmac.compare_digest(supplied_token, REINDEX_TOKEN):
+        return JSONResponse(
+            {"error": "invalid reindex credentials"},
+            status_code=401,
+            headers={"WWW-Authenticate": "Bearer", "Cache-Control": "no-store"},
+        )
     lock_file = acquire_index_lock(blocking=False)
     if lock_file is None:
         return JSONResponse({"error": "indexing already in progress"}, status_code=409)
@@ -1058,7 +1081,7 @@ def reindex():
     try:
         REINDEX_LAST_START.parent.mkdir(parents=True, exist_ok=True)
         REINDEX_LAST_START.write_text(str(now), encoding="utf-8")
-        threading.Thread(target=run_reindex, args=(lock_file,), daemon=True).start()
+        start_reindex_worker(lock_file)
     except Exception:
         lock_file.close()
         REINDEX_LAST_START.unlink(missing_ok=True)