Deploy cold-start reliability update (source: 85cf4fa)

Batch ingestion + serve-fast warmup + runtime healthcheck tuning

.gitignore

@@ -1,3 +1,5 @@
+assets/*
+!assets/.gitkeep
 dataset_build/
 dataset_build_smoke/
 assets_runtime_smoke/

Dockerfile

@@ -32,7 +32,7 @@ COPY --chown=user assets ./assets
 
 EXPOSE 7860
 
-HEALTHCHECK --interval=30s --timeout=10s --start-period=600s --retries=3 \
-    CMD curl -f http://localhost:7860/__hyperview__/health || exit 1
+HEALTHCHECK --interval=30s --timeout=10s --start-period=180s --retries=10 \
+    CMD sh -c "curl -f http://127.0.0.1:${PORT:-7860}/__hyperview__/health || exit 1"
 
 CMD ["python", "demo.py"]

README.md

@@ -27,6 +27,10 @@ Runtime environment variables:
 - `HF_DATASET_SPLIT` (default: `train`)
 - `EMBEDDING_ASSET_DIR` (default: `./assets`)
 - `EMBEDDING_ASSET_MANIFEST` (default: `${EMBEDDING_ASSET_DIR}/manifest.json`)
+- `HYPERVIEW_STARTUP_MODE` (default: `serve_fast`; choices: `serve_fast|blocking`)
+- `HYPERVIEW_WARMUP_STATUS_PATH` (default: `/tmp/hyperview_warmup_status.json`)
+- `HYPERVIEW_WARMUP_FAILURE_POLICY` (default: `exit`; choices: `exit|warn`)
+- `HYPERVIEW_BATCH_INSERT_SIZE` (default: `500`; controls sample-batch insertion chunk size)
 - `HYPERVIEW_DEFAULT_PANEL` (default: `spherical3d`; enables Sphere 3D as initial scatter panel)
 - `HYPERVIEW_LAYOUT_CACHE_VERSION` (default: `v6`; bumps dock layout localStorage key to invalidate stale cached panel state)
 - `HYPERVIEW_BIND_HOST` (preferred bind host; optional)
@@ -40,6 +44,33 @@ On Hugging Face Spaces, `SPACE_HOST` may be injected as `<space-subdomain>.hf.sp
 
 The runtime also patches HyperView's dock-layout cache key from legacy `hyperview:dockview-layout:v5` to `hyperview:dockview-layout:${HYPERVIEW_LAYOUT_CACHE_VERSION}` to force migration away from stale panel layouts after UI/layout changes. For future migrations, increment `HYPERVIEW_LAYOUT_CACHE_VERSION` (for example, `v7`) without changing code.
 
+## Startup and Warmup Semantics
+
+- `HYPERVIEW_STARTUP_MODE=serve_fast` (default):
+  - Starts the HyperView server immediately.
+  - Runs dataset warmup asynchronously in a background thread.
+  - Warmup phases are persisted as JSON: `ingest -> spaces -> layouts -> ready`.
+- `HYPERVIEW_STARTUP_MODE=blocking`:
+  - Performs warmup synchronously before serving traffic.
+
+Warmup status JSON fields include:
+
+- `status` (`starting|running|ready|failed`)
+- `phase` (`boot|ingest|spaces|layouts|ready|failed`)
+- `counts` (sample/space/layout counters and ingestion stats)
+- `error` (exception payload when warmup fails)
+- `timestamps` (`started_at`, `updated_at`, plus terminal timestamps)
+
+Failure policy behavior:
+
+- `HYPERVIEW_WARMUP_FAILURE_POLICY=exit` (default): process exits on warmup failure.
+- `HYPERVIEW_WARMUP_FAILURE_POLICY=warn`: process stays up and records failure in warmup status JSON.
+
+Healthcheck semantics:
+
+- Container health (`/__hyperview__/health`) indicates server liveness only.
+- Data readiness (dataset/spaces/layouts completed) is indicated by warmup status JSON (`status=ready`).
+
 ## Important Note
 
 HyperView similarity search currently uses cosine distance in storage backends. The Lorentz panel in this Space is intended for embedding-space visualization and geometry-aware comparison rather than canonical Lorentz-distance retrieval scoring.

demo.py

@@ -6,6 +6,10 @@ from __future__ import annotations
 import json
 import os
 import re
+import threading
+import time
+import traceback
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 
@@ -14,6 +18,7 @@ from datasets import Dataset as HFDataset
 from datasets import DatasetDict as HFDatasetDict
 from datasets import load_dataset, load_from_disk
 import hyperview as hv
+from hyperview.api import Session
 from hyperview.core.sample import Sample
 
 SPACE_HOST = os.environ.get("SPACE_HOST", "0.0.0.0")
@@ -34,6 +39,113 @@ ASSET_MANIFEST_PATH = Path(
     os.environ.get("EMBEDDING_ASSET_MANIFEST", str((EMBEDDING_ASSET_DIR / "manifest.json").resolve()))
 )
 
+DEFAULT_STARTUP_MODE = "serve_fast"
+DEFAULT_FAILURE_POLICY = "exit"
+DEFAULT_BATCH_INSERT_SIZE = 500
+DEFAULT_WARMUP_STATUS_PATH = Path("/tmp/hyperview_warmup_status.json")
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _resolve_startup_mode() -> str:
+    startup_mode = os.environ.get("HYPERVIEW_STARTUP_MODE", DEFAULT_STARTUP_MODE).strip().lower()
+    if startup_mode in {"serve_fast", "blocking"}:
+        return startup_mode
+    print(
+        f"Invalid HYPERVIEW_STARTUP_MODE={startup_mode!r}; "
+        f"falling back to {DEFAULT_STARTUP_MODE!r}."
+    )
+    return DEFAULT_STARTUP_MODE
+
+
+def _resolve_failure_policy() -> str:
+    failure_policy = os.environ.get("HYPERVIEW_WARMUP_FAILURE_POLICY", DEFAULT_FAILURE_POLICY).strip().lower()
+    if failure_policy in {"exit", "warn"}:
+        return failure_policy
+    print(
+        f"Invalid HYPERVIEW_WARMUP_FAILURE_POLICY={failure_policy!r}; "
+        f"falling back to {DEFAULT_FAILURE_POLICY!r}."
+    )
+    return DEFAULT_FAILURE_POLICY
+
+
+def _resolve_batch_insert_size() -> int:
+    raw_value = os.environ.get("HYPERVIEW_BATCH_INSERT_SIZE", str(DEFAULT_BATCH_INSERT_SIZE)).strip()
+    try:
+        batch_size = int(raw_value)
+    except ValueError as exc:
+        raise ValueError(f"Invalid integer value for HYPERVIEW_BATCH_INSERT_SIZE: {raw_value}") from exc
+    if batch_size <= 0:
+        raise ValueError(f"HYPERVIEW_BATCH_INSERT_SIZE must be > 0; got {batch_size}")
+    return batch_size
+
+
+def _resolve_warmup_status_path() -> Path:
+    raw = os.environ.get("HYPERVIEW_WARMUP_STATUS_PATH")
+    if raw is None:
+        return DEFAULT_WARMUP_STATUS_PATH
+    return Path(raw)
+
+
+class WarmupStatusTracker:
+    """Tracks warmup state and persists it to a JSON status file."""
+
+    def __init__(self, status_path: Path):
+        self._status_path = status_path
+        self._lock = threading.Lock()
+        now = _utc_now()
+        self._state: dict[str, Any] = {
+            "status": "starting",
+            "phase": "boot",
+            "counts": {},
+            "error": None,
+            "timestamps": {
+                "started_at": now,
+                "updated_at": now,
+            },
+        }
+        self._persist_locked()
+
+    def update(
+        self,
+        *,
+        status: str | None = None,
+        phase: str | None = None,
+        counts: dict[str, Any] | None = None,
+        error: dict[str, Any] | None = None,
+    ) -> None:
+        with self._lock:
+            now = _utc_now()
+            if status is not None:
+                self._state["status"] = status
+            if phase is not None:
+                self._state["phase"] = phase
+            if counts:
+                self._state["counts"].update(counts)
+            if error is not None:
+                self._state["error"] = error
+            self._state["timestamps"]["updated_at"] = now
+            if status == "ready":
+                self._state["timestamps"]["ready_at"] = now
+            if status == "failed":
+                self._state["timestamps"]["failed_at"] = now
+            self._persist_locked()
+
+    @property
+    def path(self) -> Path:
+        return self._status_path
+
+    def _persist_locked(self) -> None:
+        try:
+            self._status_path.parent.mkdir(parents=True, exist_ok=True)
+            tmp_path = self._status_path.with_name(f"{self._status_path.name}.tmp")
+            tmp_path.write_text(json.dumps(self._state, indent=2, sort_keys=True), encoding="utf-8")
+            tmp_path.replace(self._status_path)
+        except OSError as exc:
+            print(f"Warmup status warning: failed writing status JSON to {self._status_path} ({exc})")
+
 
 def _patch_hyperview_default_panel() -> None:
     """Patch HyperView 0.3.1 frontend for default panel and dock cache-key migration.
@@ -169,45 +281,103 @@ def _load_hf_rows() -> HFDataset:
     return load_dataset(HF_DATASET_REPO, name=HF_DATASET_CONFIG, split=HF_DATASET_SPLIT)
 
 
-def ingest_hf_dataset_samples(dataset: hv.Dataset) -> None:
+def ingest_hf_dataset_samples(dataset: hv.Dataset, batch_insert_size: int | None = None) -> dict[str, int]:
     rows = _load_hf_rows()
     media_root = Path(os.environ.get("HYPERVIEW_MEDIA_DIR", "./demo_data/media")) / DATASET_NAME
     media_root.mkdir(parents=True, exist_ok=True)
 
-    added = 0
+    effective_batch_size = _resolve_batch_insert_size() if batch_insert_size is None else int(batch_insert_size)
+    if effective_batch_size <= 0:
+        raise ValueError(f"batch_insert_size must be > 0; got {effective_batch_size}")
+
+    records_by_id: dict[str, dict[str, Any]] = {}
+    duplicate_ids = 0
     for index, row in enumerate(rows):
         filename = str(row.get("filename", f"sample_{index:06d}.jpg"))
         sample_id = str(row.get("sample_id", filename))
-        if dataset._storage.get_sample(sample_id) is not None:
+        if sample_id in records_by_id:
+            duplicate_ids += 1
             continue
+        records_by_id[sample_id] = {
+            "index": index,
+            "filename": filename,
+            "sample_id": sample_id,
+            "label": str(row.get("label", "")),
+            "split_tag": str(row.get("split_tag", "unknown")),
+        }
 
+    candidate_records = list(records_by_id.values())
+    candidate_ids = [record["sample_id"] for record in candidate_records]
+    existing_ids = dataset._storage.get_existing_ids(candidate_ids) if candidate_ids else set()
+    missing_records = [record for record in candidate_records if record["sample_id"] not in existing_ids]
+
+    print(
+        "HF ingestion plan: "
+        f"candidates={len(candidate_records)} existing={len(existing_ids)} "
+        f"missing={len(missing_records)} duplicates={duplicate_ids} "
+        f"batch_insert_size={effective_batch_size}"
+    )
+
+    added = 0
+    saved_images = 0
+    pending_samples: list[Sample] = []
+
+    def flush_pending_samples() -> None:
+        nonlocal added
+        if not pending_samples:
+            return
+        dataset._storage.add_samples_batch(pending_samples)
+        added += len(pending_samples)
+        print(f"Inserted sample batch: size={len(pending_samples)} total_inserted={added}")
+        pending_samples.clear()
+
+    for record in missing_records:
+        sample_id = str(record["sample_id"])
+        filename = str(record["filename"])
+
+        row = rows[int(record["index"])]
         image_obj = row["image"]
         image_path = media_root / f"{Path(sample_id).stem}.jpg"
         if not image_path.exists():
             image_obj.convert("RGB").save(image_path, format="JPEG", quality=90, optimize=True)
+            saved_images += 1
 
-        label = str(row.get("label", ""))
         metadata = {
             "filename": filename,
             "sample_id": sample_id,
-            "split_tag": str(row.get("split_tag", "unknown")),
-            "identity": label,
+            "split_tag": str(record["split_tag"]),
+            "identity": str(record["label"]),
             "source_repo": HF_DATASET_REPO,
             "source_config": HF_DATASET_CONFIG,
             "source_split": HF_DATASET_SPLIT,
         }
 
-        dataset.add_sample(
+        pending_samples.append(
             Sample(
                 id=sample_id,
                 filepath=str(image_path),
-                label=label,
+                label=str(record["label"]),
                 metadata=metadata,
             )
         )
-        added += 1
 
-    print(f"Ingested {added} HF samples into HyperView dataset '{DATASET_NAME}'.")
+        if len(pending_samples) >= effective_batch_size:
+            flush_pending_samples()
+
+    flush_pending_samples()
+
+    print(
+        f"Ingested {added} HF samples into HyperView dataset '{DATASET_NAME}' "
+        f"(saved_images={saved_images}, existing={len(existing_ids)})."
+    )
+
+    return {
+        "candidates": len(candidate_records),
+        "existing": len(existing_ids),
+        "added": added,
+        "saved_images": saved_images,
+        "duplicates": duplicate_ids,
+    }
 
 
 def ensure_embedding_spaces(dataset: hv.Dataset, asset_manifest: dict[str, Any], asset_dir: Path) -> None:
@@ -280,46 +450,200 @@ def ensure_layouts(dataset: hv.Dataset, asset_manifest: dict[str, Any]) -> list[
     return layout_keys
 
 
-def build_dataset() -> hv.Dataset:
+def _run_warmup(dataset: hv.Dataset, tracker: WarmupStatusTracker) -> None:
     asset_manifest = load_asset_manifest(ASSET_MANIFEST_PATH)
+    tracker.update(
+        status="running",
+        phase="ingest",
+        counts={"manifest_models": len(asset_manifest.get("models", []))},
+    )
 
-    dataset = hv.Dataset(DATASET_NAME)
+    batch_insert_size = _resolve_batch_insert_size()
     if len(dataset) == 0:
         print(
             f"Loading HF dataset rows from {HF_DATASET_REPO}[{HF_DATASET_CONFIG}] split={HF_DATASET_SPLIT}"
         )
-        ingest_hf_dataset_samples(dataset)
+        ingest_stats = ingest_hf_dataset_samples(dataset, batch_insert_size=batch_insert_size)
+    else:
+        ingest_stats = {
+            "candidates": len(dataset),
+            "existing": len(dataset),
+            "added": 0,
+            "saved_images": 0,
+            "duplicates": 0,
+        }
+        print(f"Skipping HF ingestion because dataset '{DATASET_NAME}' already has {len(dataset)} samples.")
 
+    tracker.update(
+        counts={
+            "batch_insert_size": batch_insert_size,
+            "dataset_samples": len(dataset),
+            **ingest_stats,
+        }
+    )
+
+    tracker.update(phase="spaces")
     ensure_embedding_spaces(dataset, asset_manifest=asset_manifest, asset_dir=EMBEDDING_ASSET_DIR)
+    tracker.update(counts={"spaces": len(dataset.list_spaces())})
+
+    tracker.update(phase="layouts")
     layout_keys = ensure_layouts(dataset, asset_manifest=asset_manifest)
 
+    tracker.update(
+        status="ready",
+        phase="ready",
+        counts={
+            "dataset_samples": len(dataset),
+            "spaces": len(dataset.list_spaces()),
+            "layouts": len(layout_keys),
+        },
+    )
+
     print(f"Dataset '{DATASET_NAME}' has {len(dataset)} samples")
     print(f"Spaces: {[space.space_key for space in dataset.list_spaces()]}")
     print(f"Layouts: {layout_keys}")
 
-    return dataset
+
+def _run_warmup_blocking(dataset: hv.Dataset, tracker: WarmupStatusTracker) -> None:
+    try:
+        _run_warmup(dataset, tracker)
+    except Exception as exc:
+        tb = traceback.format_exc()
+        tracker.update(
+            status="failed",
+            phase="failed",
+            error={
+                "type": type(exc).__name__,
+                "message": str(exc),
+                "traceback": tb,
+            },
+        )
+        print(tb)
+        raise
+
+
+def _warmup_worker(
+    dataset: hv.Dataset,
+    tracker: WarmupStatusTracker,
+    failure_policy: str,
+    failure_event: threading.Event,
+    failure_holder: dict[str, str],
+) -> None:
+    try:
+        _run_warmup(dataset, tracker)
+    except Exception as exc:
+        tb = traceback.format_exc()
+        tracker.update(
+            status="failed",
+            phase="failed",
+            error={
+                "type": type(exc).__name__,
+                "message": str(exc),
+                "traceback": tb,
+            },
+        )
+        print("Warmup failed:")
+        print(tb)
+        failure_holder["error"] = f"{type(exc).__name__}: {exc}"
+        if failure_policy == "exit":
+            failure_event.set()
+
+
+def _start_server_session(dataset: hv.Dataset, bind_host: str, bind_port: int) -> Session:
+    session = Session(dataset, host=bind_host, port=bind_port)
+    session.start(background=True)
+    print(f"HyperView server is running at {session.url}")
+    return session
+
+
+def _serve_forever(
+    session: Session,
+    *,
+    failure_event: threading.Event | None = None,
+    failure_holder: dict[str, str] | None = None,
+) -> None:
+    try:
+        while True:
+            time.sleep(0.25)
+            if session._server_thread is not None and not session._server_thread.is_alive():
+                raise RuntimeError("HyperView server stopped unexpectedly.")
+
+            if failure_event is not None and failure_event.is_set():
+                reason = None
+                if failure_holder is not None:
+                    reason = failure_holder.get("error")
+                if reason:
+                    raise RuntimeError(f"Warmup failed and failure policy is 'exit': {reason}")
+                raise RuntimeError("Warmup failed and failure policy is 'exit'.")
+    except KeyboardInterrupt:
+        pass
+    finally:
+        session.stop()
+        if session._server_thread is not None:
+            session._server_thread.join(timeout=2.0)
 
 
 def main() -> None:
     _patch_hyperview_default_panel()
-    dataset = build_dataset()
 
-    if os.environ.get("HYPERVIEW_DEMO_PREP_ONLY") == "1":
-        print("Preparation-only mode enabled; skipping server launch.")
-        return
+    startup_mode = _resolve_startup_mode()
+    failure_policy = _resolve_failure_policy()
+    warmup_status_path = _resolve_warmup_status_path()
+
+    dataset = hv.Dataset(DATASET_NAME)
+    tracker = WarmupStatusTracker(warmup_status_path)
+    tracker.update(
+        counts={
+            "dataset_samples": len(dataset),
+            "startup_mode": startup_mode,
+            "failure_policy": failure_policy,
+            "batch_insert_size": _resolve_batch_insert_size(),
+        }
+    )
 
     bind_host, bind_warning = _resolve_bind_host()
     bind_port = _resolve_port()
 
     if bind_warning:
         print(f"Bind host notice: {bind_warning}")
+
     print(
-        "Starting HyperView with "
-        f"bind_host={bind_host} bind_port={bind_port} "
+        "Starting HyperView runtime with "
+        f"startup_mode={startup_mode} failure_policy={failure_policy} "
+        f"status_path={warmup_status_path} bind_host={bind_host} bind_port={bind_port} "
         f"(SPACE_HOST={SPACE_HOST!r}, SPACE_PORT={os.environ.get('SPACE_PORT')!r}, "
         f"PORT={os.environ.get('PORT')!r})"
     )
-    hv.launch(dataset, host=bind_host, port=bind_port, open_browser=False)
+
+    if os.environ.get("HYPERVIEW_DEMO_PREP_ONLY") == "1":
+        _run_warmup_blocking(dataset, tracker)
+        print("Preparation-only mode enabled; skipping server launch.")
+        return
+
+    if startup_mode == "blocking":
+        _run_warmup_blocking(dataset, tracker)
+        session = _start_server_session(dataset, bind_host=bind_host, bind_port=bind_port)
+        _serve_forever(session)
+        return
+
+    failure_event = threading.Event()
+    failure_holder: dict[str, str] = {}
+
+    warmup_thread = threading.Thread(
+        target=_warmup_worker,
+        name="hyperview-warmup",
+        args=(dataset, tracker, failure_policy, failure_event, failure_holder),
+        daemon=True,
+    )
+    warmup_thread.start()
+    print("Warmup thread started in background.")
+
+    session = _start_server_session(dataset, bind_host=bind_host, bind_port=bind_port)
+
+    if failure_policy == "exit":
+        _serve_forever(session, failure_event=failure_event, failure_holder=failure_holder)
+    else:
+        _serve_forever(session)
 
 
 if __name__ == "__main__":