Upload 6 files

app/bucket_snapshot.py

@@ -53,14 +53,17 @@ def snapshot_status() -> dict[str, Any] | None:
         return None
 
 def publish_due(bucket_dir: Path, minimum_interval_seconds: int) -> bool:
+    return publish_delay(bucket_dir, minimum_interval_seconds) <= 0
+
+def publish_delay(bucket_dir: Path, minimum_interval_seconds: int) -> int:
     if minimum_interval_seconds <= 0:
-        return True
+        return 0
     try:
         manifest = json.loads(manifest_path(bucket_dir).read_text(encoding="utf-8"))
         created_at = int(manifest.get("created_at", 0))
     except (OSError, ValueError, TypeError, json.JSONDecodeError):
-        return True
-    return time.time() - created_at >= minimum_interval_seconds
+        return 0
+    return max(0, int(minimum_interval_seconds - (time.time() - created_at)))
 
 
 def local_repository() -> Path:
@@ -75,6 +78,12 @@ def clear_directory(path: Path) -> None:
         else:
             child.unlink()
 
+def link_or_copy(source: str | os.PathLike[str], destination: str | os.PathLike[str]) -> None:
+    try:
+        os.link(source, destination)
+    except OSError:
+        shutil.copy2(source, destination)
+
 
 def sha256(path: Path) -> str:
     digest = hashlib.sha256()
@@ -184,8 +193,40 @@ def copy_validated(source: Path, destination: Path, metadata: dict[str, Any], at
 
 
 def publish(bucket_dir: Path) -> None:
-    with serving_lock(exclusive=True):
-        _publish(bucket_dir)
+    # Readers can continue while publication snapshots one stable generation;
+    # generation activation takes the exclusive lock and waits for this copy.
+    try:
+        with serving_lock():
+            _publish(bucket_dir)
+    except Exception as exc:
+        cleanup_failed_publication(bucket_dir)
+        detail = str(exc).replace("\n", " ")[:500]
+        write_snapshot_status("publish_failed", stage="publish", reason=type(exc).__name__, detail=detail)
+        raise
+
+def cleanup_failed_publication(bucket_dir: Path) -> None:
+    keep_generation: str | None = None
+    try:
+        raw = json.loads(manifest_path(bucket_dir).read_text(encoding="utf-8"))
+        candidate = raw.get("generation") if isinstance(raw, dict) else None
+        if isinstance(candidate, str) and SNAPSHOT_NAME_RE.fullmatch(candidate):
+            keep_generation = candidate
+    except (OSError, ValueError, TypeError, json.JSONDecodeError):
+        pass
+    try:
+        generations = bucket_dir / GENERATIONS_ROOT
+        if generations.is_dir():
+            for path in generations.iterdir():
+                if path.is_dir() and SNAPSHOT_NAME_RE.fullmatch(path.name) and path.name != keep_generation:
+                    shutil.rmtree(path, ignore_errors=True)
+        for path in bucket_dir.glob(f".{MANIFEST_NAME}.*.tmp"):
+            path.unlink(missing_ok=True)
+    except OSError:
+        pass
+    try:
+        clear_directory(local_repository())
+    except OSError:
+        pass
 
 def _publish(bucket_dir: Path) -> None:
     if not bucket_dir.is_dir():
@@ -218,7 +259,7 @@ def _publish(bucket_dir: Path) -> None:
     write_snapshot_status("publishing", stage="copy_generation", snapshot=snapshot)
     generation_dir = bucket_dir / GENERATIONS_ROOT / generation
     generation_dir.mkdir(parents=True, exist_ok=False)
-    shutil.copytree(local_repository(), generation_dir / REPOSITORY_DIRNAME)
+    shutil.copytree(local_repository(), generation_dir / REPOSITORY_DIRNAME, copy_function=link_or_copy)
     sidecars: dict[str, dict[str, Any]] = {}
     with tempfile.TemporaryDirectory(dir=DATA_ROOT, prefix="bha-snapshot-") as temporary_dir:
         for key, source in SIDECARS.items():
@@ -252,7 +293,10 @@ def _publish(bucket_dir: Path) -> None:
             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())
+    try:
+        clear_directory(local_repository())
+    except OSError as exc:
+        print(f"bha_bucket_snapshot=local_repository_cleanup_failed error={type(exc).__name__}", flush=True)
     print(f"bha_bucket_snapshot=published snapshot={snapshot} elapsed_seconds={time.monotonic() - started:.2f}", flush=True)
 
 
@@ -260,6 +304,7 @@ def restore_sidecars(bucket_dir: Path, manifest: dict[str, Any]) -> None:
     generation_dir = bucket_dir / GENERATIONS_ROOT / str(manifest["generation"])
     DATA_ROOT.mkdir(parents=True, exist_ok=True)
     temporary_paths: list[tuple[Path, Path]] = []
+    replaced_targets: list[Path] = []
     try:
         for key, target in SIDECARS.items():
             metadata = manifest["sidecars"][key]
@@ -282,8 +327,9 @@ def restore_sidecars(bucket_dir: Path, manifest: dict[str, Any]) -> None:
                     Path(f"{target}-shm").unlink(missing_ok=True)
                 for temporary, target in temporary_paths:
                     os.replace(temporary, target)
+                    replaced_targets.append(target)
             except Exception:
-                for _, target in temporary_paths:
+                for target in replaced_targets:
                     target.unlink(missing_ok=True)
                 for backup, target in backups:
                     os.replace(backup, target)
@@ -332,6 +378,7 @@ def restore(bucket_dir: Path, allow_stale: bool = True) -> bool:
         write_snapshot_status("unavailable", stage="read_manifest", reason="missing_or_invalid_manifest")
         print("bha_bucket_snapshot=unavailable reason=missing_or_invalid_manifest", flush=True)
         return False
+    source_stale = False
     try:
         remote_commits = {str(key): value for key, value in parsed_archive_commits(remote=True).items()}
         if manifest["archive_commits"] != remote_commits:
@@ -339,10 +386,12 @@ def restore(bucket_dir: Path, allow_stale: bool = True) -> bool:
                 write_snapshot_status("unavailable", stage="validate_source", reason="source_fingerprint_mismatch")
                 print("bha_bucket_snapshot=unavailable reason=source_fingerprint_mismatch", flush=True)
                 return False
+            source_stale = True
             write_snapshot_status("stale_source", stage="validate_source", reason="source_fingerprint_mismatch")
             print("bha_bucket_snapshot=stale_source reason=source_fingerprint_mismatch", flush=True)
     except Exception as exc:
         if allow_stale:
+            source_stale = True
             write_snapshot_status("source_fingerprint_unverified", stage="validate_source", reason=type(exc).__name__)
             print(f"bha_bucket_snapshot=source_fingerprint_unverified reason={type(exc).__name__}", flush=True)
         else:
@@ -414,7 +463,7 @@ def restore(bucket_dir: Path, allow_stale: bool = True) -> bool:
     clear_directory(local_repository())
     write_snapshot_status(
         "restored", stage="complete", snapshot=str(manifest["snapshot"]),
-        elapsed_seconds=round(time.monotonic() - started, 2),
+        elapsed_seconds=round(time.monotonic() - started, 2), source_stale=source_stale,
     )
     print(f"bha_bucket_snapshot=restored snapshot={manifest['snapshot']} elapsed_seconds={time.monotonic() - started:.2f}", flush=True)
     return True

app/data_loader.py

@@ -5,6 +5,7 @@ import re
 import shutil
 import subprocess
 import unicodedata
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from typing import Any, Callable, Iterator, Iterable
 import jieba
@@ -14,6 +15,11 @@ CJK_RE = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff]+")
 SEARCH_RUN_RE = re.compile(r"[a-z0-9]+|[\u3400-\u4dbf\u4e00-\u9fff]+", re.IGNORECASE)
 DOCUMENT_BUILD_VERSION = "1"
 PARSED_CACHE_ROOT = Path(os.environ["BHA_PARSED_CACHE_ROOT"]) if os.environ.get("BHA_PARSED_CACHE_ROOT") else None
+try:
+    REMOTE_GIT_TIMEOUT_SECONDS = max(1, int(os.environ.get("BHA_REMOTE_GIT_TIMEOUT_SECONDS", "30")))
+except ValueError:
+    REMOTE_GIT_TIMEOUT_SECONDS = 30
+REMOTE_GIT_WORKERS = 8
 
 def initialize_search_tokenizer() -> None:
     jieba.initialize()
@@ -21,6 +27,12 @@ def initialize_search_tokenizer() -> None:
 def run(command: list[str], cwd: Path | None = None) -> None:
     subprocess.run(command, cwd=str(cwd) if cwd else None, check=True)
 
+def link_or_copy(source: str | os.PathLike[str], destination: str | os.PathLike[str]) -> None:
+    try:
+        os.link(source, destination)
+    except OSError:
+        shutil.copy2(source, destination)
+
 def cache_parsed_repository(source: Path, archive_id: int) -> None:
     if PARSED_CACHE_ROOT is None:
         return
@@ -33,7 +45,7 @@ def cache_parsed_repository(source: Path, archive_id: int) -> None:
             return
     temporary = PARSED_CACHE_ROOT / f".archives{archive_id}.{os.getpid()}.tmp"
     shutil.rmtree(temporary, ignore_errors=True)
-    shutil.copytree(source, temporary)
+    shutil.copytree(source, temporary, copy_function=link_or_copy)
     shutil.rmtree(target, ignore_errors=True)
     temporary.replace(target)
 
@@ -82,7 +94,7 @@ def ensure_parsed_data(
             shutil.rmtree(target)
         cached = PARSED_CACHE_ROOT / f"archives{archive_id}" if PARSED_CACHE_ROOT is not None else None
         if cached is not None and cached.is_dir() and (cached / ".git").exists():
-            shutil.copytree(cached, target)
+            shutil.copytree(cached, target, copy_function=link_or_copy)
         else:
             run(["git", "clone", "--depth", "1", "--branch", "parsed", repo, str(target)])
         expected = expected_commits.get(archive_id) if expected_commits else None
@@ -125,17 +137,9 @@ def parsed_archive_revisions(remote: bool = False, archive_ids: Iterable[int] |
 
 def parsed_archive_commits(remote: bool = False, archive_ids: Iterable[int] | None = None) -> dict[int, str]:
     revisions: dict[int, str] = {}
-    selected = archive_ids if archive_ids is not None else range(ARCHIVE_START, ARCHIVE_END + 1)
-    for archive_id in selected:
-        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:
+    selected = list(archive_ids) if archive_ids is not None else list(range(ARCHIVE_START, ARCHIVE_END + 1))
+    if not remote:
+        for archive_id in selected:
             result = subprocess.run(
                 ["git", "rev-parse", "HEAD"],
                 cwd=PARSED_ROOT / f"archives{archive_id}",
@@ -144,9 +148,27 @@ def parsed_archive_commits(remote: bool = False, archive_ids: Iterable[int] | No
                 text=True,
             )
             revision = result.stdout.strip()
+            if not re.fullmatch(r"[0-9a-f]{40}", revision):
+                raise RuntimeError(f"invalid parsed commit for archive {archive_id}")
+            revisions[archive_id] = revision
+        return revisions
+
+    def fetch_remote(archive_id: int) -> tuple[int, str]:
+        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,
+            timeout=REMOTE_GIT_TIMEOUT_SECONDS,
+        )
+        revision = result.stdout.split(maxsplit=1)[0]
         if not re.fullmatch(r"[0-9a-f]{40}", revision):
             raise RuntimeError(f"invalid parsed commit for archive {archive_id}")
-        revisions[archive_id] = revision
+        return archive_id, revision
+
+    with ThreadPoolExecutor(max_workers=min(REMOTE_GIT_WORKERS, max(1, len(selected)))) as executor:
+        for archive_id, revision in executor.map(fetch_remote, selected):
+            revisions[archive_id] = revision
     return revisions
 
 def read_json(path: Path) -> Any:
@@ -330,3 +352,17 @@ def iter_documents(
                     "path": str(article_path.relative_to(archive_root)),
                     "article": {**article, "tags": tags},
                 }
+
+def iter_document_ids(archive_ids: Iterable[int] | None = None) -> Iterator[str]:
+    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
+        for metadata_path in archive_root.glob("*/*/*.metadata"):
+            publication_id = metadata_path.stem
+            for article_path in metadata_path.parent.glob("*/*.json"):
+                article_id = article_path.stem
+                yield f"{archive_id}:{len(article_id)}:{article_id}:{publication_id}"

app/doc_store.py

@@ -40,6 +40,7 @@ def init_db(conn: sqlite3.Connection) -> None:
         );
         CREATE INDEX IF NOT EXISTS idx_docs_title ON docs(title);
         CREATE INDEX IF NOT EXISTS idx_docs_publication_name ON docs(publication_name);
+        CREATE INDEX IF NOT EXISTS idx_docs_archive_id ON docs(archive_id);
         """
     )
     columns = {row[1] for row in conn.execute("PRAGMA table_info(docs)")}

app/facet_store.py

@@ -30,6 +30,7 @@ def init_db(conn: sqlite3.Connection) -> None:
             PRIMARY KEY (kind, name)
         );
         CREATE INDEX IF NOT EXISTS idx_facets_kind_name ON facets(kind, name);
+        CREATE INDEX IF NOT EXISTS idx_facets_kind_name_nocase ON facets(kind, name COLLATE NOCASE);
         """
     )
 

app/indexer.py

@@ -13,7 +13,7 @@ from typing import IO, Iterable
 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 DOCUMENT_BUILD_VERSION, ensure_parsed_data, iter_documents, parsed_archive_commits, parsed_archive_revisions, parsed_corpus_fingerprint
+from .data_loader import DOCUMENT_BUILD_VERSION, ensure_parsed_data, iter_document_ids, iter_documents, parsed_archive_commits, parsed_archive_revisions, parsed_corpus_fingerprint
 from .doc_store import DOC_DB, DOC_STORE_SCHEMA_VERSION, connect as connect_doc_db, delete_docs_to_conn, doc_row, init_db as init_doc_db, reset_db, row_to_doc, update_years_to_conn, write_docs_to_conn
 from .facet_store import FACET_DB, FACET_STORE_SCHEMA_VERSION, apply_facet_deltas, 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
@@ -238,16 +238,30 @@ def incremental_build(
             raise IncrementalUnsafe("serving sidecar generation is unavailable")
         removed_docs: list[dict] = []
         upsert_docs: list[dict] = []
+        scan_archives = changed
+        if changed_paths is not None:
+            # A changed-path filter cannot identify deletions without seeing the
+            # complete current archive. Empty diffs are safe to skip.
+            scan_archives = [
+                archive_id for archive_id in changed
+                if changed_paths.get(archive_id) is None or changed_paths[archive_id]
+            ]
         with connect_doc_db(temp_doc_db) as conn:
             init_doc_db(conn)
-            placeholders = ",".join("?" for _ in changed)
-            remaining_ids = {
-                row[0] for row in conn.execute(
-                    f"SELECT doc_id FROM docs WHERE archive_id IN ({placeholders})", changed
-                )
-            }
+            remaining_ids: set[str] = set()
+            if scan_archives:
+                placeholders = ",".join("?" for _ in scan_archives)
+                old_ids = {
+                    row[0] for row in conn.execute(
+                        f"SELECT doc_id FROM docs WHERE archive_id IN ({placeholders})", scan_archives
+                    )
+                }
+                current_ids = set(iter_document_ids(scan_archives))
+                remaining_ids = old_ids - current_ids
             seen_ids: set[str] = set()
-            for doc in iter_documents(changed, changed_paths=changed_paths):
+            # File paths identify deletions cheaply; only changed files need
+            # JSON normalization and tokenization.
+            for doc in iter_documents(scan_archives, changed_paths=changed_paths):
                 doc_id = doc["doc_id"]
                 if doc_id in seen_ids:
                     raise IncrementalUnsafe(f"duplicate incremental doc_id: {doc_id}")
@@ -682,6 +696,12 @@ def _ensure_index(
         if error_count:
             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")
+        with connect_doc_db(temp_doc_db) as conn:
+            sqlite_count = int(conn.execute("SELECT COUNT(*) FROM docs").fetchone()[0])
+        if sqlite_count != ok:
+            raise RuntimeError(
+                f"full build document count mismatch: indexed={ok} sqlite={sqlite_count}; duplicate doc_id likely"
+            )
         write_facets(facet_counters, temp_facet_db)
         es.index(index=build_index, id="__meta__", document=desired_metadata(source_fingerprint, archive_revisions))
         with connect_doc_db(temp_doc_db) as conn:

app/main.py

@@ -12,6 +12,7 @@ import socket
 import tempfile
 import threading
 import time
+import uuid
 from urllib.parse import quote, unquote_to_bytes, urlparse
 from typing import Any, Literal
 from zipfile import ZIP_DEFLATED, ZipFile
@@ -30,8 +31,8 @@ from .doc_store import available_years, get_doc, random_doc, variants_for_title,
 from .search_store import SearchStoreError, get_search_docs, literal_match_ids
 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, publish_due as bucket_snapshot_publish_due, snapshot_status as bucket_snapshot_status
+from .indexer import acquire_index_lock, ensure_index, sidecars_ready
+from .bucket_snapshot import publish as publish_bucket_snapshot, publish_delay as bucket_snapshot_publish_delay, publish_due as bucket_snapshot_publish_due, snapshot_status as bucket_snapshot_status, write_snapshot_status
 from .storage_lock import serving_lock
 SOURCE_MIRROR_MANIFEST_URL = os.environ.get(
     "BHA_SOURCE_MIRROR_MANIFEST_URL",
@@ -47,9 +48,12 @@ source_mirror_loaded_at = 0.0
 source_mirror_etag: str | None = None
 source_mirror_refresh_task: asyncio.Task | None = None
 source_mirror_lock = asyncio.Lock()
+snapshot_publish_timer: threading.Timer | None = None
+snapshot_publish_timer_lock = threading.Lock()
 MAX_RESULT_WINDOW = 500000
 REINDEX_COOLDOWN_SECONDS = 10 * 60
 REINDEX_LAST_START = DATA_ROOT / "reindex-last-start"
+REINDEX_JOB_PATH = DATA_ROOT / "reindex-job.json"
 REINDEX_TOKEN = os.environ.get("BHA_REINDEX_TOKEN", "")
 PROOFREAD_GITHUB_TOKEN = os.environ.get("BHA_PROOFREAD_GITHUB_TOKEN", "")
 PROOFREAD_PIPELINE_REPOSITORY = os.environ.get("BHA_PROOFREAD_PIPELINE_REPOSITORY", "anftm/pipeline")
@@ -540,6 +544,32 @@ def clear_response_caches() -> None:
     sources_response_cache = None
     status_response_cache = None
 
+def write_reindex_job(state: str, job_id: str, **values: Any) -> None:
+    previous = read_reindex_job() or {}
+    payload = {
+        **(previous if previous.get("job_id") == job_id else {}),
+        "job_id": job_id, "state": state, "updated_at": int(time.time()), **values,
+    }
+    REINDEX_JOB_PATH.parent.mkdir(parents=True, exist_ok=True)
+    temporary = REINDEX_JOB_PATH.with_name(f".{REINDEX_JOB_PATH.name}.{uuid.uuid4().hex}.tmp")
+    temporary.write_text(json.dumps(payload, ensure_ascii=False, sort_keys=True), encoding="utf-8")
+    os.replace(temporary, REINDEX_JOB_PATH)
+
+def read_reindex_job() -> dict[str, Any] | None:
+    try:
+        value = json.loads(REINDEX_JOB_PATH.read_text(encoding="utf-8"))
+        return value if isinstance(value, dict) else None
+    except (OSError, json.JSONDecodeError):
+        return None
+
+def update_reindex_job(state: str, job_id: str, **values: Any) -> None:
+    if not job_id:
+        return
+    try:
+        write_reindex_job(state, job_id, **values)
+    except OSError as exc:
+        print(f"reindex_job_status_failed={type(exc).__name__}", flush=True)
+
 def search_cache_key(body: SearchRequest, page: int, page_size: int) -> tuple[Any, ...]:
     return (
         body.q, page, page_size, body.exact, body.fulltext, body.sort,
@@ -623,6 +653,7 @@ async def initialize_when_ready() -> None:
 
 async def start_ready_data_initialization():
     await asyncio.to_thread(initialize_search_tokenizer)
+    resume_pending_snapshot_publish()
     asyncio.create_task(initialize_when_ready())
     asyncio.create_task(auxiliary_warmup_loop())
     asyncio.create_task(source_mirror_warmup_loop())
@@ -686,6 +717,13 @@ def index_ready() -> bool:
     except Exception:
         return False
 
+def serving_generation_valid() -> bool:
+    try:
+        metadata = es.get(index=INDEX_NAME, id="__meta__").get("_source", {})
+        return bool(metadata) and sidecars_ready(metadata)
+    except Exception:
+        return False
+
 def document_count() -> int:
     try:
         if not es.indices.exists(index=INDEX_NAME):
@@ -1842,14 +1880,53 @@ async def download_source_zip(doc_id: str):
     )
 
 def publish_bucket_snapshot_worker() -> None:
+    global snapshot_publish_timer
+    with snapshot_publish_timer_lock:
+        snapshot_publish_timer = None
     try:
-        if BUCKET_DIR is not None and bucket_snapshot_publish_due(BUCKET_DIR, SNAPSHOT_MIN_INTERVAL_SECONDS):
+        if BUCKET_DIR is None:
+            return
+        if bucket_snapshot_publish_due(BUCKET_DIR, SNAPSHOT_MIN_INTERVAL_SECONDS):
             publish_bucket_snapshot(BUCKET_DIR)
-        elif BUCKET_DIR is not None:
-            print("bucket_snapshot_publish_skipped=minimum_interval", flush=True)
+            return
+        delay = bucket_snapshot_publish_delay(BUCKET_DIR, SNAPSHOT_MIN_INTERVAL_SECONDS)
+        write_snapshot_status("pending_publish", retry_at=int(time.time()) + delay)
+        print(f"bucket_snapshot_publish_scheduled=delay_{delay}", flush=True)
+        schedule_bucket_snapshot_publish(delay)
     except Exception as exc:
+        if BUCKET_DIR is not None:
+            retry_delay = 300
+            try:
+                write_snapshot_status(
+                    "pending_publish", retry_at=int(time.time()) + retry_delay,
+                    last_error=f"{type(exc).__name__}: {str(exc)[:300]}",
+                )
+                schedule_bucket_snapshot_publish(retry_delay)
+            except Exception:
+                pass
         print(f"bucket_snapshot_publish_failed={exc}")
 
+def schedule_bucket_snapshot_publish(delay: int) -> None:
+    global snapshot_publish_timer
+    with snapshot_publish_timer_lock:
+        if snapshot_publish_timer is not None and snapshot_publish_timer.is_alive():
+            return
+        snapshot_publish_timer = threading.Timer(max(1, delay), publish_bucket_snapshot_worker)
+        snapshot_publish_timer.daemon = True
+        snapshot_publish_timer.start()
+
+def resume_pending_snapshot_publish() -> None:
+    if BUCKET_DIR is None:
+        return
+    status = bucket_snapshot_status() or {}
+    if status.get("state") != "pending_publish":
+        return
+    try:
+        delay = max(0, int(status.get("retry_at", 0)) - int(time.time()))
+    except (TypeError, ValueError):
+        delay = 0
+    schedule_bucket_snapshot_publish(delay)
+
 def run_reindex(
     lock_file,
     reset: bool = False,
@@ -1857,7 +1934,10 @@ def run_reindex(
     expected_revisions: dict[int, str] | None = None,
     changed_archives: list[int] | None = None,
 ) -> None:
+    job = read_reindex_job()
+    job_id = str(job.get("job_id")) if job and job.get("job_id") else ""
     try:
+        update_reindex_job("running", job_id, started_at=int(time.time()))
         INDEX_STATUS_PATH.write_text("rebuilding", encoding="utf-8")
         clear_response_caches()
         if expected_commits is None and expected_revisions is None:
@@ -1869,11 +1949,18 @@ def run_reindex(
                 changed_archives=changed_archives,
             )
         INDEX_STATUS_PATH.write_text("ready", encoding="utf-8")
+        update_reindex_job(
+            "succeeded", job_id, changed=bool(changed), index_status="ready", finished_at=int(time.time()),
+        )
         if changed and BUCKET_DIR is not None:
             threading.Thread(target=publish_bucket_snapshot_worker, daemon=True).start()
     except Exception as exc:
-        safe_ready = not INDEX_SWITCH_STATE_PATH.exists() and index_ready()
+        safe_ready = not INDEX_SWITCH_STATE_PATH.exists() and serving_generation_valid()
         INDEX_STATUS_PATH.write_text("ready" if safe_ready else "failed", encoding="utf-8")
+        update_reindex_job(
+            "failed", job_id, error=f"{type(exc).__name__}: {str(exc)[:500]}",
+            index_status="ready" if safe_ready else "failed", finished_at=int(time.time()),
+        )
         print(f"reindex_failed={exc}")
     finally:
         clear_response_caches()
@@ -1935,6 +2022,13 @@ def reindex(request: Request, full: bool = False, payload: ReindexRequest | None
     try:
         REINDEX_LAST_START.parent.mkdir(parents=True, exist_ok=True)
         REINDEX_LAST_START.write_text(str(now), encoding="utf-8")
+        job_id = uuid.uuid4().hex
+        write_reindex_job(
+            "accepted", job_id, mode="full" if full else "incremental",
+            changed_archives=changed_archives or [],
+            target_archive_revisions=expected_revisions or {},
+            accepted_at=now,
+        )
         if payload is None:
             start_reindex_worker(lock_file, reset=full)
         else:
@@ -1951,6 +2045,7 @@ def reindex(request: Request, full: bool = False, payload: ReindexRequest | None
         "ok": True,
         "status": "accepted",
         "mode": "full" if full else "incremental",
+        "job_id": job_id,
         "cooldown_seconds": REINDEX_COOLDOWN_SECONDS,
     }
 
@@ -1967,6 +2062,7 @@ def reindex_status(request: Request):
     return {
         "index_status": index_status(),
         "progress": index_progress(),
+        "job": read_reindex_job(),
         "switch_pending": INDEX_SWITCH_STATE_PATH.exists(),
         "archive_revisions": metadata.get("archive_revisions") or {},
         "build_fingerprint": metadata.get("build_fingerprint"),