Upload 3 files

app/bucket_snapshot.py

@@ -8,7 +8,9 @@ import os
 import re
 import shutil
 import sqlite3
+import subprocess
 import sys
+import tempfile
 import time
 import uuid
 from pathlib import Path
@@ -16,9 +18,10 @@ from typing import Any
 
 from elasticsearch import Elasticsearch
 
-from .config import ARCHIVE_END, ARCHIVE_START, DATA_ROOT, ES_URL, INDEX_NAME, INDEX_VERSION
+from .config import ARCHIVE_END, ARCHIVE_START, DATA_ROOT, ES_URL, INDEX_NAME, INDEX_VERSION, REPO_PREFIX
 from .doc_store import DOC_DB
 from .facet_store import FACET_DB
+from .storage_lock import serving_lock
 
 MANIFEST_NAME = "current.json"
 REPOSITORY = "bha_bucket"
@@ -33,6 +36,23 @@ def expected_metadata() -> dict[str, Any]:
     return {"version": INDEX_VERSION, "archive_start": ARCHIVE_START, "archive_end": ARCHIVE_END}
 
 
+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]
+        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()
+
+
 def sha256(path: Path) -> str:
     digest = hashlib.sha256()
     with path.open("rb") as source:
@@ -66,11 +86,17 @@ def read_manifest(bucket_dir: Path) -> dict[str, Any] | None:
         return None
     snapshot = data.get("snapshot")
     generation = data.get("generation")
+    physical_index = data.get("physical_index")
+    source_fingerprint = data.get("source_fingerprint")
     sidecars = data.get("sidecars")
     if not isinstance(snapshot, str) or not SNAPSHOT_NAME_RE.fullmatch(snapshot):
         return None
     if not isinstance(generation, str) or not SNAPSHOT_NAME_RE.fullmatch(generation):
         return None
+    if not isinstance(physical_index, str) or not physical_index.startswith(f"{INDEX_NAME}-"):
+        return None
+    if not isinstance(source_fingerprint, str) or not re.fullmatch(r"[0-9a-f]{64}", source_fingerprint):
+        return None
     if not isinstance(sidecars, dict) or set(sidecars) != set(SIDECARS):
         return None
     for key, target in SIDECARS.items():
@@ -80,6 +106,19 @@ def read_manifest(bucket_dir: Path) -> dict[str, Any] | None:
     return data
 
 
+def physical_index(client: Elasticsearch) -> str:
+    aliases = client.indices.get_alias(name=INDEX_NAME)
+    names = sorted(aliases)
+    if len(names) != 1 or not names[0].startswith(f"{INDEX_NAME}-"):
+        raise RuntimeError("expected one physical index behind the article alias")
+    return names[0]
+
+
+def backup_sqlite(source: Path, destination: Path) -> None:
+    with sqlite3.connect(source) as source_conn, sqlite3.connect(destination) as destination_conn:
+        source_conn.backup(destination_conn)
+
+
 def publish(bucket_dir: Path) -> None:
     if not bucket_dir.is_dir():
         raise RuntimeError(f"bucket directory is unavailable: {bucket_dir}")
@@ -89,6 +128,8 @@ def publish(bucket_dir: Path) -> None:
     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)
+    physical_name = physical_index(client)
     snapshot = f"bha-{int(time.time())}-{uuid.uuid4().hex[:8]}"
     generation = snapshot
     repository(client, bucket_dir)
@@ -96,22 +137,27 @@ def publish(bucket_dir: Path) -> None:
     client.snapshot.create(
         repository=REPOSITORY,
         snapshot=snapshot,
-        indices=INDEX_NAME,
+        indices=physical_name,
         include_global_state=False,
         wait_for_completion=True,
     )
     generation_dir = bucket_dir / GENERATIONS_ROOT / generation
     generation_dir.mkdir(parents=True, exist_ok=False)
     sidecars: dict[str, dict[str, Any]] = {}
-    for key, source in SIDECARS.items():
-        destination = generation_dir / source.name
-        shutil.copyfile(source, destination)
-        sidecars[key] = file_metadata(destination)
+    with tempfile.TemporaryDirectory(dir=DATA_ROOT, prefix="bha-snapshot-") as temporary_dir:
+        for key, source in SIDECARS.items():
+            local_copy = Path(temporary_dir) / source.name
+            backup_sqlite(source, local_copy)
+            destination = generation_dir / source.name
+            shutil.copyfile(local_copy, destination)
+            sidecars[key] = file_metadata(destination)
     payload = {
         "schema": 1,
         "index_name": INDEX_NAME,
         "index_version": INDEX_VERSION,
         "index_metadata": expected_metadata(),
+        "source_fingerprint": source_fingerprint,
+        "physical_index": physical_name,
         "snapshot": snapshot,
         "generation": generation,
         "document_count": int(client.count(index=INDEX_NAME, query={"exists": {"field": "doc_id"}})["count"]),
@@ -121,6 +167,17 @@ def publish(bucket_dir: Path) -> None:
     temporary_manifest = bucket_dir / f".{MANIFEST_NAME}.{uuid.uuid4().hex}.tmp"
     temporary_manifest.write_text(json.dumps(payload, ensure_ascii=False, sort_keys=True), encoding="utf-8")
     os.replace(temporary_manifest, manifest_path(bucket_dir))
+    try:
+        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)
+    except Exception as exc:
+        print(f"bha_bucket_snapshot=cleanup_failed error={type(exc).__name__}", flush=True)
     print(f"bha_bucket_snapshot=published snapshot={snapshot} elapsed_seconds={time.monotonic() - started:.2f}", flush=True)
 
 
@@ -137,8 +194,24 @@ def restore_sidecars(bucket_dir: Path, manifest: dict[str, Any]) -> None:
             temporary = DATA_ROOT / f".{target.name}.{uuid.uuid4().hex}.tmp"
             shutil.copyfile(source, temporary)
             temporary_paths.append((temporary, target))
-        for temporary, target in temporary_paths:
-            os.replace(temporary, target)
+        with serving_lock(exclusive=True):
+            backups: list[tuple[Path, Path]] = []
+            try:
+                for _, target in temporary_paths:
+                    if target.exists():
+                        backup = DATA_ROOT / f".{target.name}.{uuid.uuid4().hex}.previous"
+                        os.replace(target, backup)
+                        backups.append((backup, target))
+                for temporary, target in temporary_paths:
+                    os.replace(temporary, target)
+            except Exception:
+                for _, target in temporary_paths:
+                    target.unlink(missing_ok=True)
+                for backup, target in backups:
+                    os.replace(backup, target)
+                raise
+            for backup, _ in backups:
+                backup.unlink(missing_ok=True)
     finally:
         for temporary, _ in temporary_paths:
             temporary.unlink(missing_ok=True)
@@ -164,6 +237,13 @@ def restore(bucket_dir: Path) -> bool:
     if manifest is None:
         print("bha_bucket_snapshot=unavailable reason=missing_or_invalid_manifest", flush=True)
         return False
+    try:
+        if manifest["source_fingerprint"] != 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
     client = Elasticsearch(ES_URL, request_timeout=3600)
     if client.indices.exists(index=INDEX_NAME):
         print("bha_bucket_snapshot=skipped reason=index_exists", flush=True)
@@ -174,7 +254,7 @@ def restore(bucket_dir: Path) -> bool:
         client.snapshot.restore(
             repository=REPOSITORY,
             snapshot=str(manifest["snapshot"]),
-            indices=INDEX_NAME,
+            indices=str(manifest["physical_index"]),
             include_global_state=False,
             include_aliases=True,
             wait_for_completion=True,

app/indexer.py

@@ -87,7 +87,7 @@ def index_mapping() -> dict:
                 "date_year_months": {"type": "keyword"},
                 "date_year_days": {"type": "keyword"},
                 "date_year_month_days": {"type": "keyword"},
-                "content": {"type": "text", "analyzer": "cjk"},
+                "content": {"type": "text", "analyzer": "cjk", "index_phrases": True},
                 "path": {"type": "keyword"},
             },
         },

app/main.py

@@ -3,6 +3,7 @@ from html import escape
 import ipaddress
 import asyncio
 import json
+import os
 import random
 import shutil
 import socket
@@ -27,11 +28,13 @@ from .doc_store import available_years, get_doc, get_search_docs, random_doc, va
 from .data_loader import initialize_search_tokenizer, search_words
 from .facet_store import FACET_DB, list_facets, sources_payload
 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_LAST_START = DATA_ROOT / "reindex-last-start"
 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")
 app.add_middleware(
     CORSMiddleware,
@@ -1024,6 +1027,8 @@ def run_reindex(lock_file) -> None:
         INDEX_STATUS_PATH.write_text("rebuilding", encoding="utf-8")
         clear_response_caches()
         ensure_index(reset=True, lock_file=lock_file)
+        if BUCKET_DIR is not None:
+            publish_bucket_snapshot(BUCKET_DIR)
         INDEX_STATUS_PATH.write_text("ready", encoding="utf-8")
     except Exception as exc:
         INDEX_STATUS_PATH.write_text("ready" if index_ready() else "failed", encoding="utf-8")