fix suggestion rebuild flow and es index creation

- consolidate suggestion rebuild flow into build_suggestions.sh via --rebuild and remove the redundant rebuild_suggestions.sh wrapper - make suggestion versioned index names use microseconds and handle index-create retries/timeouts without false already_exists failures - treat create requests as successful when the index was created server-side, then explicitly wait for shard readiness and surface allocation diagnostics - clean up freshly created suggestion indices on rebuild failure to avoid leaving red orphan indices behind - make rebuild smoke tests target the local backend by default, with SUGGESTIONS_SMOKE_BASE_URL as the explicit override - add unit coverage for microsecond versioned index names and cleanup on unallocatable index failures

fix suggestion rebuild flow and es index creation
- consolidate suggestion rebuild flow into build_suggestions.sh via --rebuild and remove the redundant rebuild_suggestions.sh wrapper - make suggestion versioned index names use microseconds and handle index-create retries/timeouts without false already_exists failures - treat create requests as successful when the index was created server-side, then explicitly wait for shard readiness and surface allocation diagnostics - clean up freshly created suggestion indices on rebuild failure to avoid leaving red orphan indices behind - make rebuild smoke tests target the local backend by default, with SUGGESTIONS_SMOKE_BASE_URL as the explicit override - add unit coverage for microsecond versioned index names and cleanup on unallocatable index failures
tangwang
1 parent e50924ed
Showing 6 changed files with 275 additions and 74 deletions Show diff stats
scripts/build_suggestions.sh
scripts/rebuild_suggestions.sh
scripts/service_ctl.sh
suggestion/builder.py
tests/test_suggestions.py
utils/es_client.py
@@ -10,34 +10,34 @@
 #   ./scripts/build_suggestions.sh <tenant_id> --mode incremental
 #
 #   # full rebuild + incremental + ES/API smoke checks (same as legacy rebuild_suggestions.sh)
-#   ./scripts/build_suggestions.sh <tenant_id> --rebuild-and-verify
+#   ./scripts/build_suggestions.sh <tenant_id> --rebuild
 #
 set -euo pipefail
 if [ $# -lt 1 ]; then
-  echo "Usage: $0 <tenant_id> [--rebuild-and-verify | extra args for main.py build-suggestions...]"
-  echo "Example (full): $0 162 --mode full --days 30 --publish-alias"
-  echo "Example (incremental): $0 162 --mode incremental --overlap-minutes 30"
-  echo "Example (pipeline + smoke): $0 162 --rebuild-and-verify"
+  echo "Usage: $0 <tenant_id> [--rebuild | extra args for main.py build-suggestions...]"
+  echo "Example (full): $0 163 --mode full --days 30 --publish-alias"
+  echo "Example (incremental): $0 163 --mode incremental --overlap-minutes 30"
+  echo "Example (pipeline + smoke): $0 163 --rebuild"
   exit 1
 fi
 TENANT_ID="$1"
 shift || true
-REBUILD_VERIFY=false
+RUN_REBUILD_PIPELINE=false
 PASSTHROUGH_ARGS=()
 for arg in "$@"; do
-  if [ "$arg" = "--rebuild-and-verify" ]; then
-    REBUILD_VERIFY=true
+  if [ "$arg" = "--rebuild" ]; then
+    RUN_REBUILD_PIPELINE=true
   else
     PASSTHROUGH_ARGS+=("$arg")
   fi
 done
-if [ "$REBUILD_VERIFY" = true ] && [ ${#PASSTHROUGH_ARGS[@]} -gt 0 ]; then
-  echo "Error: --rebuild-and-verify cannot be combined with other build-suggestions arguments."
+if [ "$RUN_REBUILD_PIPELINE" = true ] && [ ${#PASSTHROUGH_ARGS[@]} -gt 0 ]; then
+  echo "Error: --rebuild cannot be combined with other build-suggestions arguments."
   exit 1
 fi
@@ -50,11 +50,12 @@ if [ ! -x &quot;$PY_BIN&quot; ]; then
   PY_BIN="python3"
 fi
-if [ "$REBUILD_VERIFY" = true ]; then
-  # Fixed smoke-test queries and languages (no CLI args).
+if [ "$RUN_REBUILD_PIPELINE" = true ]; then
   SAMPLE_QUERIES=(s sh dress tshirt)
   SAMPLE_LANGS=(en zh)
-  API_BASE="${API_BASE_URL:-http://localhost:6002}"
+  # This script validates the locally rebuilt index, so default the smoke target
+  # to the local backend. A remote/public API base must be opted into explicitly.
+  API_BASE="${SUGGESTIONS_SMOKE_BASE_URL:-http://localhost:6002}"
   if [ -z "${ES_HOST:-}" ]; then
     ES_HOST="$("$PY_BIN" - <<'PY'
@@ -111,7 +112,7 @@ PY
 }'
   echo
-  echo "[4/4] API smoke test"
+  echo "[4/4] API smoke test (base=${API_BASE})"
   for lang in "${SAMPLE_LANGS[@]}"; do
     for q in "${SAMPLE_QUERIES[@]}"; do
       echo "--- GET /search/suggestions?q=${q}&language=${lang} ---"
@@ -1,6 +0,0 @@
-#!/usr/bin/env bash
-# Backward-compatible entry: full + incremental + ES/API checks.
-# See scripts/build_suggestions.sh --rebuild-and-verify.
-set -euo pipefail
-SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-exec "$SCRIPT_DIR/build_suggestions.sh" "${1:?Usage: $0 <tenant_id>}" --rebuild-and-verify
@@ -16,9 +16,10 @@ mkdir -p &quot;${LOG_DIR}&quot;
 source "${PROJECT_ROOT}/scripts/lib/load_env.sh"
 CORE_SERVICES=("backend" "indexer" "frontend" "eval-web")
-OPTIONAL_SERVICES=("tei" "cnclip" "embedding" "embedding-image" "translator" "reranker" "reranker-fine")
+# reranker-fine 暂时不用，因此暂时从OPTIONAL_SERVICES中删除
+OPTIONAL_SERVICES=("tei" "cnclip" "embedding" "embedding-image" "translator" "reranker")
 FULL_SERVICES=("${OPTIONAL_SERVICES[@]}" "${CORE_SERVICES[@]}")
-STOP_ORDER_SERVICES=("frontend" "eval-web" "indexer" "backend" "reranker-fine" "reranker" "translator" "embedding-image" "embedding" "cnclip" "tei")
+STOP_ORDER_SERVICES=("frontend" "eval-web" "indexer" "backend" "reranker" "translator" "embedding-image" "embedding" "cnclip" "tei")
 all_services() {
   echo "${FULL_SERVICES[@]}"
@@ -38,7 +38,7 @@ def get_suggestion_alias_name(tenant_id: str) -&gt; str:
 def get_suggestion_versioned_index_name(tenant_id: str, build_at: Optional[datetime] = None) -> str:
     """Versioned suggestion index name."""
-    ts = (build_at or datetime.now(timezone.utc)).strftime("%Y%m%d%H%M%S")
+    ts = (build_at or datetime.now(timezone.utc)).strftime("%Y%m%d%H%M%S%f")
     return f"{_index_prefix()}search_suggestions_tenant_{tenant_id}_v{ts}"
@@ -101,6 +101,79 @@ class SuggestionIndexBuilder:
         self.es_client = es_client
         self.db_engine = db_engine
+    def _format_allocation_failure(self, index_name: str) -> str:
+        health = self.es_client.wait_for_index_ready(index_name=index_name, timeout="5s")
+        explain = self.es_client.get_allocation_explain(index_name=index_name)
+
+        parts = [
+            f"Suggestion index '{index_name}' was created but is not allocatable/readable yet",
+            f"health_status={health.get('status')}",
+            f"timed_out={health.get('timed_out')}",
+        ]
+        if health.get("error"):
+            parts.append(f"health_error={health['error']}")
+
+        if explain:
+            unassigned = explain.get("unassigned_info") or {}
+            if unassigned.get("reason"):
+                parts.append(f"unassigned_reason={unassigned['reason']}")
+            if unassigned.get("last_allocation_status"):
+                parts.append(f"last_allocation_status={unassigned['last_allocation_status']}")
+
+            for node in explain.get("node_allocation_decisions") or []:
+                node_name = node.get("node_name") or node.get("node_id") or "unknown-node"
+                for decider in node.get("deciders") or []:
+                    if decider.get("decision") == "NO":
+                        parts.append(
+                            f"{node_name}:{decider.get('decider')}={decider.get('explanation')}"
+                        )
+                        return "; ".join(parts)
+
+        return "; ".join(parts)
+
+    def _create_fresh_versioned_index(
+        self,
+        tenant_id: str,
+        mapping: Dict[str, Any],
+        max_attempts: int = 5,
+    ) -> str:
+        for attempt in range(1, max_attempts + 1):
+            index_name = get_suggestion_versioned_index_name(tenant_id)
+            if self.es_client.index_exists(index_name):
+                logger.warning(
+                    "Suggestion index name collision before create for tenant=%s index=%s attempt=%s/%s",
+                    tenant_id,
+                    index_name,
+                    attempt,
+                    max_attempts,
+                )
+                continue
+
+            if self.es_client.create_index(index_name, mapping):
+                return index_name
+
+            if self.es_client.index_exists(index_name):
+                logger.warning(
+                    "Suggestion index name collision during create for tenant=%s index=%s attempt=%s/%s",
+                    tenant_id,
+                    index_name,
+                    attempt,
+                    max_attempts,
+                )
+                continue
+
+            raise RuntimeError(f"Failed to create suggestion index: {index_name}")
+
+        raise RuntimeError(
+            f"Failed to allocate a unique suggestion index name for tenant={tenant_id} after {max_attempts} attempts"
+        )
+
+    def _ensure_new_index_ready(self, index_name: str) -> None:
+        health = self.es_client.wait_for_index_ready(index_name=index_name, timeout="5s")
+        if health.get("ok"):
+            return
+        raise RuntimeError(self._format_allocation_failure(index_name))
+
     @staticmethod
     def _to_utc(dt: Any) -> Optional[datetime]:
         if dt is None:
@@ -609,62 +682,65 @@ class SuggestionIndexBuilder:
         index_languages: List[str] = tenant_cfg.get("index_languages") or ["en", "zh"]
         primary_language: str = tenant_cfg.get("primary_language") or "en"
-        # Always write to a fresh versioned index; legacy concrete index is no longer supported.
-        index_name = get_suggestion_versioned_index_name(tenant_id)
-
-        if self.es_client.index_exists(index_name):
-            raise RuntimeError(f"Target suggestion index already exists: {index_name}")
-
-        mapping = build_suggestion_mapping(index_languages=index_languages)
-        if not self.es_client.create_index(index_name, mapping):
-            raise RuntimeError(f"Failed to create suggestion index: {index_name}")
+        alias_publish: Optional[Dict[str, Any]] = None
+        index_name: Optional[str] = None
+        try:
+            mapping = build_suggestion_mapping(index_languages=index_languages)
+            index_name = self._create_fresh_versioned_index(
+                tenant_id=tenant_id,
+                mapping=mapping,
+            )
+            self._ensure_new_index_ready(index_name)
-        key_to_candidate = self._build_full_candidates(
-            tenant_id=tenant_id,
-            index_languages=index_languages,
-            primary_language=primary_language,
-            days=days,
-            batch_size=batch_size,
-            min_query_len=min_query_len,
-        )
+            key_to_candidate = self._build_full_candidates(
+                tenant_id=tenant_id,
+                index_languages=index_languages,
+                primary_language=primary_language,
+                days=days,
+                batch_size=batch_size,
+                min_query_len=min_query_len,
+            )
-        now_iso = datetime.now(timezone.utc).isoformat()
-        docs = [self._candidate_to_doc(tenant_id, c, now_iso) for c in key_to_candidate.values()]
+            now_iso = datetime.now(timezone.utc).isoformat()
+            docs = [self._candidate_to_doc(tenant_id, c, now_iso) for c in key_to_candidate.values()]
-        if docs:
-            bulk_result = self.es_client.bulk_index(index_name=index_name, docs=docs)
-            self.es_client.refresh(index_name)
-        else:
-            bulk_result = {"success": 0, "failed": 0, "errors": []}
+            if docs:
+                bulk_result = self.es_client.bulk_index(index_name=index_name, docs=docs)
+                self.es_client.refresh(index_name)
+            else:
+                bulk_result = {"success": 0, "failed": 0, "errors": []}
-        alias_publish: Optional[Dict[str, Any]] = None
-        if publish_alias:
-            alias_publish = self._publish_alias(
-                tenant_id=tenant_id,
-                index_name=index_name,
-                keep_versions=keep_versions,
-            )
+            if publish_alias:
+                alias_publish = self._publish_alias(
+                    tenant_id=tenant_id,
+                    index_name=index_name,
+                    keep_versions=keep_versions,
+                )
-        now_utc = datetime.now(timezone.utc).isoformat()
-        meta_patch: Dict[str, Any] = {
-            "last_full_build_at": now_utc,
-            "last_incremental_watermark": now_utc,
-        }
-        if publish_alias:
-            meta_patch["active_index"] = index_name
-            meta_patch["active_alias"] = get_suggestion_alias_name(tenant_id)
-        self._upsert_meta(tenant_id, meta_patch)
+            now_utc = datetime.now(timezone.utc).isoformat()
+            meta_patch: Dict[str, Any] = {
+                "last_full_build_at": now_utc,
+                "last_incremental_watermark": now_utc,
+            }
+            if publish_alias:
+                meta_patch["active_index"] = index_name
+                meta_patch["active_alias"] = get_suggestion_alias_name(tenant_id)
+            self._upsert_meta(tenant_id, meta_patch)
-        return {
-            "mode": "full",
-            "tenant_id": str(tenant_id),
-            "index_name": index_name,
-            "alias_published": bool(alias_publish),
-            "alias_publish": alias_publish,
-            "total_candidates": len(key_to_candidate),
-            "indexed_docs": len(docs),
-            "bulk_result": bulk_result,
-        }
+            return {
+                "mode": "full",
+                "tenant_id": str(tenant_id),
+                "index_name": index_name,
+                "alias_published": bool(alias_publish),
+                "alias_publish": alias_publish,
+                "total_candidates": len(key_to_candidate),
+                "indexed_docs": len(docs),
+                "bulk_result": bulk_result,
+            }
+        except Exception:
+            if index_name and not alias_publish:
+                self.es_client.delete_index(index_name)
+            raise
     def _build_incremental_deltas(
         self,
@@ -8,6 +8,7 @@ from suggestion.builder import (
     QueryDelta,
     SuggestionIndexBuilder,
     get_suggestion_alias_name,
+    get_suggestion_versioned_index_name,
 )
 from suggestion.service import SuggestionService
@@ -121,6 +122,16 @@ class FakeESClient:
         self.indices.add(index_name)
         return True
+    def wait_for_index_ready(self, index_name: str, timeout: str = "10s") -> Dict[str, Any]:
+        self.calls.append({"op": "wait_for_index_ready", "index": index_name, "timeout": timeout})
+        return {"ok": True, "status": "green", "timed_out": False}
+
+    def get_allocation_explain(self, index_name: str, shard: int = 0, primary: bool = True) -> Dict[str, Any] | None:
+        self.calls.append(
+            {"op": "get_allocation_explain", "index": index_name, "shard": shard, "primary": primary}
+        )
+        return None
+
     def refresh(self, index_name: str) -> bool:
         self.calls.append({"op": "refresh", "index": index_name})
         return True
@@ -150,6 +161,67 @@ class FakeESClient:
 @pytest.mark.unit
+def test_versioned_index_name_uses_microseconds():
+    build_at = datetime(2026, 4, 7, 3, 52, 26, 123456, tzinfo=timezone.utc)
+    assert (
+        get_suggestion_versioned_index_name("163", build_at)
+        == "search_suggestions_tenant_163_v20260407035226123456"
+    )
+
+
+@pytest.mark.unit
+def test_rebuild_cleans_up_unallocatable_new_index():
+    fake_es = FakeESClient()
+
+    def _wait_fail(index_name: str, timeout: str = "10s") -> Dict[str, Any]:
+        fake_es.calls.append({"op": "wait_for_index_ready", "index": index_name, "timeout": timeout})
+        return {"ok": False, "status": "red", "timed_out": True}
+
+    def _allocation_explain(index_name: str, shard: int = 0, primary: bool = True) -> Dict[str, Any]:
+        fake_es.calls.append(
+            {"op": "get_allocation_explain", "index": index_name, "shard": shard, "primary": primary}
+        )
+        return {
+            "unassigned_info": {"reason": "INDEX_CREATED", "last_allocation_status": "no"},
+            "node_allocation_decisions": [
+                {
+                    "node_name": "node-1",
+                    "deciders": [
+                        {
+                            "decider": "disk_threshold",
+                            "decision": "NO",
+                            "explanation": "node is above high watermark",
+                        }
+                    ],
+                }
+            ],
+        }
+
+    fake_es.wait_for_index_ready = _wait_fail  # type: ignore[method-assign]
+    fake_es.get_allocation_explain = _allocation_explain  # type: ignore[method-assign]
+
+    builder = SuggestionIndexBuilder(es_client=fake_es, db_engine=None)
+
+    from config import tenant_config_loader as tcl
+
+    loader = tcl.get_tenant_config_loader()
+    loader._config = {
+        "default": {"primary_language": "en", "index_languages": ["en", "zh"]},
+        "tenants": {
+            "163": {"primary_language": "en", "index_languages": ["en", "zh"]},
+        },
+    }
+
+    with pytest.raises(RuntimeError, match="disk_threshold"):
+        builder.rebuild_tenant_index(tenant_id="163")
+
+    create_calls = [x for x in fake_es.calls if x.get("op") == "create_index"]
+    assert len(create_calls) == 1
+    created_index = create_calls[0]["index"]
+    assert created_index not in fake_es.indices
+
+
+@pytest.mark.unit
 def test_resolve_query_language_prefers_log_field():
     fake_es = FakeESClient()
     builder = SuggestionIndexBuilder(es_client=fake_es, db_engine=None)
@@ -76,13 +76,70 @@ class ESClient:
             True if successful, False otherwise
         """
         try:
-            self.client.indices.create(index=index_name, body=body)
+            client = self.client.options(request_timeout=30, max_retries=0)
+            client.indices.create(
+                index=index_name,
+                body=body,
+                wait_for_active_shards="0",
+            )
             logger.info(f"Index '{index_name}' created successfully")
             return True
         except Exception as e:
+            if self.index_exists(index_name):
+                logger.warning(
+                    "Create index request for '%s' raised %s, but the index now exists; treating it as created",
+                    index_name,
+                    type(e).__name__,
+                    exc_info=True,
+                )
+                return True
             logger.error(f"Failed to create index '{index_name}': {e}", exc_info=True)
             return False
+    def wait_for_index_ready(self, index_name: str, timeout: str = "10s") -> Dict[str, Any]:
+        """Wait until an index primary shard is allocated and searchable."""
+        try:
+            resp = self.client.cluster.health(
+                index=index_name,
+                wait_for_status="yellow",
+                timeout=timeout,
+                level="indices",
+            )
+            index_info = ((resp.get("indices") or {}).get(index_name) or {})
+            status = index_info.get("status") or resp.get("status")
+            timed_out = bool(resp.get("timed_out"))
+            return {
+                "ok": (not timed_out) and status in {"yellow", "green"},
+                "status": status,
+                "timed_out": timed_out,
+                "response": resp,
+            }
+        except Exception as e:
+            logger.error("Failed waiting for index '%s' readiness: %s", index_name, e, exc_info=True)
+            return {
+                "ok": False,
+                "status": "unknown",
+                "timed_out": False,
+                "error": str(e),
+            }
+
+    def get_allocation_explain(self, index_name: str, shard: int = 0, primary: bool = True) -> Optional[Dict[str, Any]]:
+        """Explain why a shard can or cannot be allocated."""
+        try:
+            return self.client.cluster.allocation_explain(
+                body={"index": index_name, "shard": shard, "primary": primary}
+            )
+        except Exception as e:
+            logger.warning(
+                "Failed to get allocation explain for index '%s' shard=%s primary=%s: %s",
+                index_name,
+                shard,
+                primary,
+                e,
+                exc_info=True,
+            )
+            return None
+
     def put_alias(self, index_name: str, alias_name: str) -> bool:
         """Add alias for an index."""
         try: