Upload 6 files

app/bucket_snapshot.py

@@ -17,8 +17,8 @@ from typing import Any
 
 from elasticsearch import Elasticsearch
 
-from .config import DATA_ROOT, ES_URL, INDEX_NAME
-from .data_loader import parsed_corpus_fingerprint
+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
@@ -104,6 +104,7 @@ def read_manifest(bucket_dir: Path) -> dict[str, Any] | None:
     generation = data.get("generation")
     physical_index = data.get("physical_index")
     source_fingerprint = index_metadata.get("source_fingerprint")
+    raw_revisions = index_metadata.get("archive_revisions")
     sidecars = data.get("sidecars")
     if not isinstance(snapshot, str) or not SNAPSHOT_NAME_RE.fullmatch(snapshot):
         return None
@@ -113,7 +114,17 @@ 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):
+    if not isinstance(raw_revisions, dict):
+        return None
+    try:
+        revisions = {int(key): str(value) for key, value in raw_revisions.items()}
+    except (TypeError, ValueError):
+        return None
+    if set(revisions) != set(range(ARCHIVE_START, ARCHIVE_END + 1)):
+        return None
+    if any(not re.fullmatch(r"[0-9a-f]{40}", revision) for revision in revisions.values()):
+        return None
+    if index_metadata != desired_metadata(source_fingerprint, revisions):
         return None
     if not isinstance(sidecars, dict) or set(sidecars) != set(SIDECARS):
         return None
@@ -164,8 +175,9 @@ def publish(bucket_dir: Path) -> None:
         raise RuntimeError("cannot snapshot missing SQLite sidecars")
     client = Elasticsearch(ES_URL, request_timeout=3600)
     source_fingerprint = parsed_corpus_fingerprint()
+    revisions = parsed_archive_revisions()
     metadata = client.get(index=INDEX_NAME, id="__meta__").get("_source", {})
-    if metadata != desired_metadata(source_fingerprint):
+    if metadata != desired_metadata(source_fingerprint, revisions):
         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]}"

app/config.py

@@ -6,7 +6,6 @@ PARSED_ROOT = DATA_ROOT / "parsed"
 INDEX_NAME = os.environ.get("ES_INDEX", "article")
 ES_URL = os.environ.get("ES_URL", "http://127.0.0.1:9200")
 REPO_PREFIX = os.environ.get("REPO_PREFIX", "https://github.com/anftm")
-SOURCE_REPO_OWNER = os.environ.get("SOURCE_REPO_OWNER", "anftm")
 ARCHIVE_START = int(os.environ.get("ARCHIVE_START", "0"))
 ARCHIVE_END = int(os.environ.get("ARCHIVE_END", "31"))
 RESET_INDEX = os.environ.get("RESET_INDEX", "0") == "1"

app/data_loader.py

@@ -5,7 +5,7 @@ import shutil
 import subprocess
 import unicodedata
 from pathlib import Path
-from typing import Any, Callable, Iterator
+from typing import Any, Callable, Iterator, Iterable
 import jieba
 from .config import ARCHIVE_END, ARCHIVE_START, PARSED_ROOT, REPO_PREFIX
 
@@ -37,8 +37,16 @@ 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] = []
+def parsed_corpus_fingerprint(
+    remote: bool = False, revisions: dict[int, str] | None = None
+) -> str:
+    revisions = revisions or parsed_archive_revisions(remote=remote)
+    return hashlib.sha256(
+        "\n".join(f"{archive_id}:{revisions[archive_id]}" for archive_id in sorted(revisions)).encode()
+    ).hexdigest()
+
+def parsed_archive_revisions(remote: bool = False) -> dict[int, str]:
+    revisions: dict[int, str] = {}
     for archive_id in range(ARCHIVE_START, ARCHIVE_END + 1):
         if remote:
             result = subprocess.run(
@@ -59,8 +67,8 @@ def parsed_corpus_fingerprint(remote: bool = False) -> str:
             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()
+        revisions[archive_id] = revision
+    return revisions
 
 def read_json(path: Path) -> Any:
     try:
@@ -168,8 +176,11 @@ def normalize_tags(tags: Any) -> tuple[list[str], list[str]]:
             types.append(str(tag_type).strip())
     return names, types
 
-def iter_documents() -> Iterator[dict[str, Any]]:
-    for archive_id in range(ARCHIVE_START, ARCHIVE_END + 1):
+def iter_documents(archive_ids: Iterable[int] | None = None) -> Iterator[dict[str, Any]]:
+    selected_archives = (
+        sorted(set(archive_ids)) if archive_ids is not None else range(ARCHIVE_START, ARCHIVE_END + 1)
+    )
+    for archive_id in selected_archives:
         archive_root = PARSED_ROOT / f"archives{archive_id}"
         if not archive_root.exists():
             continue

app/doc_store.py

@@ -74,6 +74,25 @@ def write_docs_to_conn(conn: sqlite3.Connection, docs: list[dict[str, Any]]) ->
             year_rows,
         )
 
+def delete_docs_to_conn(conn: sqlite3.Connection, doc_ids: list[str]) -> None:
+    if not doc_ids:
+        return
+    placeholders = ",".join("?" for _ in doc_ids)
+    conn.execute(f"DELETE FROM docs WHERE doc_id IN ({placeholders})", doc_ids)
+
+def rebuild_years_to_conn(conn: sqlite3.Connection) -> None:
+    years: set[int] = set()
+    for (raw_dates,) in conn.execute("SELECT date_display_json FROM docs"):
+        try:
+            dates = json.loads(raw_dates or "[]")
+        except (TypeError, ValueError):
+            dates = []
+        for value in dates:
+            if isinstance(value, str) and value[:4].isdigit():
+                years.add(int(value[:4]))
+    conn.execute("DELETE FROM years")
+    conn.executemany("INSERT INTO years(year) VALUES (?)", [(year,) for year in sorted(years)])
+
 def years_from_docs(docs: list[dict[str, Any]]) -> set[int]:
     years: set[int] = set()
     for doc in docs:

app/indexer.py

@@ -12,8 +12,8 @@ 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, 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 .data_loader import ensure_parsed_data, iter_documents, parsed_archive_revisions, parsed_corpus_fingerprint
+from .doc_store import DOC_DB, connect as connect_doc_db, delete_docs_to_conn, init_db as init_doc_db, reset_db, rebuild_years_to_conn, row_to_doc, 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
 MAX_RESULT_WINDOW = 500000
@@ -106,7 +106,7 @@ def build_fingerprint() -> str:
         digest.update(module_path.read_bytes())
     return digest.hexdigest()
 
-def desired_metadata(source_fingerprint: str | None = None) -> dict:
+def desired_metadata(source_fingerprint: str | None = None, archive_revisions: dict[int, str] | None = None) -> dict:
     metadata = {
         "version": INDEX_VERSION,
         "archive_start": ARCHIVE_START,
@@ -115,15 +115,134 @@ def desired_metadata(source_fingerprint: str | None = None) -> dict:
     }
     if source_fingerprint is not None:
         metadata["source_fingerprint"] = source_fingerprint
+    if archive_revisions is not None:
+        metadata["archive_revisions"] = {str(k): v for k, v in sorted(archive_revisions.items())}
     return metadata
 
-def metadata_matches(es: Elasticsearch, source_fingerprint: str) -> bool:
+def metadata_matches(es: Elasticsearch, source_fingerprint: str, archive_revisions: dict[int, str] | None = None) -> bool:
     try:
         data = es.get(index=INDEX_NAME, id="__meta__")
     except Exception:
         return False
     source = data.get("_source", {})
-    return source == desired_metadata(source_fingerprint)
+    return source == desired_metadata(source_fingerprint, archive_revisions)
+
+class IncrementalUnsafe(RuntimeError):
+    pass
+
+def copy_sqlite(source: Path, target: Path) -> None:
+    target.unlink(missing_ok=True)
+    with sqlite3.connect(source) as source_conn, sqlite3.connect(target) as target_conn:
+        source_conn.backup(target_conn)
+
+def serving_physical_index(es: Elasticsearch) -> str | None:
+    try:
+        aliases = es.indices.get_alias(name=INDEX_NAME)
+    except Exception:
+        return None
+    indices = [name for name, data in aliases.items() if INDEX_NAME in data.get("aliases", {})]
+    return indices[0] if len(indices) == 1 else None
+
+def clone_serving_index(es: Elasticsearch, source: str, target: str) -> None:
+    try:
+        es.indices.put_settings(index=source, settings={"index.blocks.write": True})
+        es.indices.clone(
+            index=source,
+            target=target,
+            body={"settings": {"index.blocks.write": False}},
+        )
+    finally:
+        es.indices.put_settings(index=source, settings={"index.blocks.write": False})
+
+def rebuild_facets_from_docs(conn: sqlite3.Connection, target: Path) -> None:
+    counters: dict[str, Counter[str]] = {
+        "source": Counter(), "author": Counter(), "tag": Counter(),
+        "type": Counter(), "archive": Counter(),
+    }
+    for row in conn.execute("SELECT * FROM docs"):
+        update_facets(counters, row_to_doc(row))
+    write_facets(counters, target)
+
+def incremental_build(es: Elasticsearch, source_fingerprint: str, revisions: dict[int, str], old_meta: dict) -> None:
+    source_index = serving_physical_index(es)
+    old_revisions = old_meta.get("archive_revisions")
+    expected_static = desired_metadata()
+    if (
+        not source_index
+        or not isinstance(old_revisions, dict)
+        or set(old_revisions) != {str(archive_id) for archive_id in revisions}
+        or any(old_meta.get(key) != value for key, value in expected_static.items())
+    ):
+        raise IncrementalUnsafe("incremental state is unavailable")
+    changed = [
+        archive_id for archive_id in revisions
+        if old_revisions.get(str(archive_id)) != revisions[archive_id]
+    ]
+    if not changed:
+        raise IncrementalUnsafe("source fingerprint changed without an archive revision change")
+
+    generation = uuid.uuid4().hex[:12]
+    build_index = f"{INDEX_NAME}-{generation}"
+    temp_doc_db = DOC_DB.with_name(f"{DOC_DB.name}.{generation}.tmp")
+    temp_facet_db = FACET_DB.with_name(f"{FACET_DB.name}.{generation}.tmp")
+    activated = False
+    try:
+        copy_sqlite(DOC_DB, temp_doc_db)
+        copy_sqlite(FACET_DB, temp_facet_db)
+        old_docs: dict[str, dict] = {}
+        new_docs: dict[str, dict] = {}
+        with connect_doc_db(temp_doc_db) as conn:
+            init_doc_db(conn)
+            placeholders = ",".join("?" for _ in changed)
+            for row in conn.execute(f"SELECT * FROM docs WHERE archive_id IN ({placeholders})", changed):
+                old_docs[row["doc_id"]] = row_to_doc(row)
+            new_docs = {doc["doc_id"]: doc for doc in iter_documents(changed)}
+            deleted = sorted(set(old_docs) - set(new_docs))
+            delete_docs_to_conn(conn, deleted)
+            write_docs_to_conn(conn, list(new_docs.values()))
+            rebuild_years_to_conn(conn)
+            conn.commit()
+        with connect_doc_db(temp_doc_db) as conn:
+            rebuild_facets_from_docs(conn, temp_facet_db)
+
+        clone_serving_index(es, source_index, build_index)
+        actions = []
+        for doc_id in sorted(set(old_docs) - set(new_docs)):
+            actions.append({"_op_type": "delete", "_index": build_index, "_id": doc_id})
+        for doc in new_docs.values():
+            actions.append({"_op_type": "index", "_index": build_index, "_id": doc["doc_id"], "_source": doc})
+        errors = [item for ok, item in helpers.streaming_bulk(
+            es.options(request_timeout=120), actions, chunk_size=BULK_CHUNK_SIZE,
+            max_chunk_bytes=BULK_MAX_BYTES, raise_on_error=False,
+            raise_on_exception=False, retry_on_status=(429, 502, 503, 504),
+        ) if not ok]
+        if errors:
+            raise IncrementalUnsafe(f"incremental bulk update failed: {errors[0]}")
+        es.index(index=build_index, id="__meta__", document=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])
+        es_count = int(es.count(index=build_index, query={"exists": {"field": "doc_id"}}).get("count", -1))
+        if es_count != sqlite_count:
+            raise IncrementalUnsafe(
+                f"incremental document count mismatch: elasticsearch={es_count} sqlite={sqlite_count}"
+            )
+        INDEX_STATUS_PATH.write_text("switching", encoding="utf-8")
+        activate_build(es, build_index, temp_doc_db, temp_facet_db)
+        activated = True
+        INDEX_STATUS_PATH.write_text("ready", encoding="utf-8")
+    except Exception as exc:
+        uncertain = activation_uncertain(es, build_index)
+        if uncertain or isinstance(exc, IncrementalUnsafe):
+            raise
+        INDEX_STATUS_PATH.write_text("ready", encoding="utf-8")
+        raise IncrementalUnsafe(f"incremental build failed safely: {exc}") from exc
+    finally:
+        uncertain = activation_uncertain(es, build_index)
+        if not activated and not uncertain:
+            es.indices.delete(index=build_index, ignore_unavailable=True)
+            reset_db(temp_doc_db)
+            reset_facet_db(temp_facet_db)
 
 def sidecars_ready() -> bool:
     if not DOC_DB.exists() or not FACET_DB.exists():
@@ -188,6 +307,14 @@ def alias_points_to(es: Elasticsearch, index_name: str) -> bool:
         return False
     return INDEX_NAME in aliases.get(index_name, {}).get("aliases", {})
 
+def activation_uncertain(es: Elasticsearch, index_name: str) -> bool:
+    if SWITCH_STATE_PATH.exists():
+        return True
+    try:
+        return alias_points_to(es, index_name)
+    except Exception:
+        return False
+
 def recover_interrupted_switch(es: Elasticsearch) -> None:
     if not SWITCH_STATE_PATH.exists():
         return
@@ -313,14 +440,28 @@ def _ensure_index(reset: bool = False) -> None:
         archive_total=total,
         elapsed_seconds=int(time.time() - started_at),
     ))
-    source_fingerprint = parsed_corpus_fingerprint()
+    archive_revisions = parsed_archive_revisions()
+    source_fingerprint = parsed_corpus_fingerprint(revisions=archive_revisions)
     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, source_fingerprint) and sidecars_ready():
+        if not force_rebuild and count and metadata_matches(es, source_fingerprint, archive_revisions) and sidecars_ready():
             write_progress(status="ready", indexed=count, errors=0, elapsed_seconds=int(time.time() - started_at))
             return
+    if not force_rebuild and has_serving_data:
+        try:
+            old_meta = es.get(index=INDEX_NAME, id="__meta__").get("_source", {})
+            incremental_build(es, source_fingerprint, archive_revisions, old_meta)
+            write_progress(
+                status="ready",
+                indexed=es.count(index=INDEX_NAME).get("count", 0),
+                errors=0,
+                elapsed_seconds=int(time.time() - started_at),
+            )
+            return
+        except IncrementalUnsafe as exc:
+            print(f"incremental_index_unavailable={exc}; using full rebuild")
     progressive = not has_serving_data
     generation = uuid.uuid4().hex[:12]
     build_index = f"{INDEX_NAME}-{generation}"
@@ -402,7 +543,7 @@ def _ensure_index(reset: bool = False) -> None:
             write_progress(status="failed", indexed=ok, errors=error_count, elapsed_seconds=int(time.time() - started_at))
             raise RuntimeError(f"bulk indexing failed with {error_count} errors")
         write_facets(facet_counters, temp_facet_db)
-        es.index(index=build_index, id="__meta__", document=desired_metadata(source_fingerprint))
+        es.index(index=build_index, id="__meta__", document=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

@@ -11,7 +11,7 @@ import socket
 import tempfile
 import threading
 import time
-from urllib.parse import quote, urlparse, urlunparse
+from urllib.parse import quote, urlparse
 from typing import Any
 from zipfile import ZIP_DEFLATED, ZipFile
 import httpx
@@ -24,7 +24,7 @@ from brotli_asgi import BrotliMiddleware
 from starlette.background import BackgroundTask
 from fastapi.staticfiles import StaticFiles
 from pydantic import BaseModel, Field
-from .config import APP_ROOT, DATA_ROOT, ES_URL, INDEX_NAME, SOURCE_REPO_OWNER
+from .config import APP_ROOT, DATA_ROOT, ES_URL, INDEX_NAME
 from .doc_store import available_years, get_doc, random_doc, variants_for_title, warmup_db
 from .search_store import SearchStoreError, get_search_docs, literal_match_ids
 from .data_loader import initialize_search_tokenizer, search_words
@@ -831,14 +831,6 @@ def validate_source_url(url: str) -> str | None:
         return "source URL IP is not allowed"
     return None
 
-def mirror_source_url(url: str) -> str:
-    parsed = urlparse(str(url))
-    parts = parsed.path.split("/")
-    if parsed.hostname in {"github.com", "raw.githubusercontent.com"} and len(parts) > 2 and parts[1] == "banned-historical-archives":
-        parts[1] = SOURCE_REPO_OWNER
-        return urlunparse(parsed._replace(path="/".join(parts)))
-    return str(url)
-
 async def validate_source_target(url: str) -> str | None:
     if error := validate_source_url(url):
         return error
@@ -925,7 +917,7 @@ async def download_source(doc_id: str):
     files = source.get("source_files") or []
     if not files:
         return JSONResponse({"error": "source file not available"}, status_code=404)
-    url = mirror_source_url(files[0])
+    url = str(files[0])
     suffix = Path(url.split("?", 1)[0]).suffix or ".bin"
     filename = safe_filename(source_filename_base(source, doc_id), suffix)
     try:
@@ -964,8 +956,7 @@ async def download_source_zip(doc_id: str):
     downloaded: list[tuple[Path, str]] = []
     total_bytes = 0
     try:
-        for index, source_url in enumerate(files, start=1):
-            url = mirror_source_url(source_url)
+        for index, url in enumerate(files, start=1):
             source_path = temp_dir / f"source-{index:03d}"
             size = await download_source_to_path(url, source_path, MAX_SOURCE_ZIP_BYTES - total_bytes)
             total_bytes += size
@@ -996,11 +987,11 @@ async def download_source_zip(doc_id: str):
         background=BackgroundTask(shutil.rmtree, temp_dir, ignore_errors=True),
     )
 
-def run_reindex(lock_file) -> None:
+def run_reindex(lock_file, reset: bool = False) -> None:
     try:
         INDEX_STATUS_PATH.write_text("rebuilding", encoding="utf-8")
         clear_response_caches()
-        ensure_index(reset=True, lock_file=lock_file)
+        ensure_index(reset=reset, lock_file=lock_file)
         if BUCKET_DIR is not None:
             publish_bucket_snapshot(BUCKET_DIR)
         INDEX_STATUS_PATH.write_text("ready", encoding="utf-8")
@@ -1012,12 +1003,12 @@ def run_reindex(lock_file) -> None:
         clear_response_caches()
         lock_file.close()
 
-def start_reindex_worker(lock_file) -> None:
-    threading.Thread(target=run_reindex, args=(lock_file,), daemon=True).start()
+def start_reindex_worker(lock_file, reset: bool = False) -> None:
+    threading.Thread(target=run_reindex, args=(lock_file, reset), daemon=True).start()
 
 @app.post("/api/reindex", status_code=202)
 
-def reindex(request: Request):
+def reindex(request: Request, full: bool = False):
     auth_error = admin_auth_error(request)
     if auth_error is not None:
         return auth_error
@@ -1040,12 +1031,17 @@ def reindex(request: Request):
     try:
         REINDEX_LAST_START.parent.mkdir(parents=True, exist_ok=True)
         REINDEX_LAST_START.write_text(str(now), encoding="utf-8")
-        start_reindex_worker(lock_file)
+        start_reindex_worker(lock_file, reset=full)
     except Exception:
         lock_file.close()
         REINDEX_LAST_START.unlink(missing_ok=True)
         raise
-    return {"ok": True, "status": "accepted", "cooldown_seconds": REINDEX_COOLDOWN_SECONDS}
+    return {
+        "ok": True,
+        "status": "accepted",
+        "mode": "full" if full else "incremental",
+        "cooldown_seconds": REINDEX_COOLDOWN_SECONDS,
+    }
 static_dir = APP_ROOT / "static"
 app.mount("/static", StaticFiles(directory=static_dir), name="static")
 @app.get("/")