fix: persist A 改良版 (verify + 5s debounce + sync push for upload)

app/services/ingestion.py

@@ -41,7 +41,7 @@ from app.services.chunking import (
 from app.services.embedding import get_embedder
 from app.services.parsers import parse_with_fallback
 from app.services.parsers.base_parser import ParsedDocument
-from app.services.persist import schedule_push
+from app.services.persist import push_to_hf, schedule_push
 from app.services.vector_store import upsert_chunks
 
 logger = logging.getLogger(__name__)
@@ -261,8 +261,18 @@ async def _ingest_locked(content: bytes, filename: str) -> IngestResult:
             filename, doc_id, len(chunking.children), elapsed_ms,
         )
 
-        # 11. 调度持久化推送
-        await schedule_push()
+        # 11. ✅ A 改良版: 重要写操作 (upload) 同步阻塞 push, 不再吃 debounce 黑洞
+        # 用户已经等了 1-3 分钟摄入, 再加 5-15s push 可接受
+        # push 内部已带 verify, 失败会写 _state['last_error'] 让 /readyz 暴露
+        push_ok = await push_to_hf()
+        if not push_ok:
+            # push 失败但不阻塞 ingest 结果 (用户至少能看到 doc ready)
+            logger.warning(
+                "Ingest done but persist push failed for %s. "
+                "See /readyz persist.last_error. "
+                "Data is still on local /data but may not survive Space restart.",
+                doc_id,
+            )
 
         return IngestResult(
             doc_id=doc_id,
@@ -308,6 +318,8 @@ async def delete_document(doc_id: str) -> bool:
             shutil.rmtree(up_dir, ignore_errors=True)
 
     await schedule_push()
+    # ⚠️ 注意: delete 仍走 schedule_push (debounce), 不阻塞用户响应.
+    # 因为 delete 操作不产生新数据, 延迟 5s 推完全可接受.
     return True
 
 

app/services/persist.py

@@ -3,18 +3,22 @@
 为什么需要: HF Spaces 免费版磁盘是临时的 (容器重启后 /data 内的非持久卷数据会丢).
 唯一免费的持久化方案是把 /data 同步到 HF Dataset repo (Git LFS).
 
-调用模式:
-- lifespan startup: restore_from_hf()
-- 每次写操作后: schedule_push() (异步, 不阻塞用户)
-- lifespan shutdown: push_to_hf() (同步, 尽量)
-
-健壮性:
-- 首次部署 (repo 不存在) → 捕获 RepositoryNotFoundError → 标记 "fresh_start"
-- HF_TOKEN 缺失 → 跳过持久化, 降级为纯本地
-- 网络错误 → 重试 3 次后放弃, 不阻塞业务
-- upload_folder 跑在**独立 executor** 上, 不会占业务池 (否则 HF Space → HF Dataset
-  一旦卡/慢, ChromaDB query / BGE-M3 encode 等业务的 run_in_executor 全排不上, chat 直接挂)
-- push 失败时**也重置** pending_push flag, 不然 schedule_push 永远 short-circuit
+调用模式 (A 改良版):
+- lifespan startup:  restore_from_hf()          (拉 snapshot)
+- 重要写操作后:     await push_to_hf()         (同步阻塞 + verify, upload/delete 用)
+- 低频写操作:       schedule_push()            (5s debounce, chat 等用)
+- lifespan shutdown: await push_to_hf()         (兜底)
+
+健壮性 (A 改良版):
+- ✅ push 后 verify: list_repo_files 确认数据真的上去了
+- ✅ push 失败/超时状态写入 _state, /readyz 暴露给前端
+- ✅ debounce 30s → 5s (缩小"重启即丢"窗口)
+- ✅ 重要操作 (upload/delete) 改同步 push, 不再吃 debounce 黑洞
+- ✅ 独立 ThreadPoolExecutor, 不跟 Chroma / BGE-M3 业务抢线程
+- ✅ _background_tasks 持有 task 引用, 防 asyncio GC
+- ✅ push 失败也重置 pending_push, 不卡死后续写
+- 首次部署 (repo 不存在) → RepositoryNotFoundError → fresh_start
+- HF_TOKEN 缺失 → disabled 降级
 """
 from __future__ import annotations
 
@@ -22,11 +26,13 @@ import asyncio
 import concurrent.futures
 import logging
 import shutil
+import time
 from pathlib import Path
 from typing import Literal
 
 from huggingface_hub import (
     create_repo,
+    list_repo_files,
     snapshot_download,
     upload_folder,
 )
@@ -45,11 +51,21 @@ _persist_executor = concurrent.futures.ThreadPoolExecutor(
     thread_name_prefix="persist",
 )
 
-# 状态机: 持久化是否启用 / 启动模式
-_state: dict[str, str | bool] = {
-    "mode": "disabled",  # disabled | cold_restore | fresh_start
-    "last_push_at": 0.0,
-    "pending_push": False,
+# ✅ 持有后台 task 引用, 避免 asyncio GC 回收未完成的 debounce 任务
+# (曾因为 _state["pending_push"] = True 但 _delayed_push 被 GC, 导致永远卡死)
+_background_tasks: set[asyncio.Task] = set()
+
+# debounce 间隔 (秒): 5s 足够让同一秒内的多次写合并, 又不会留太久黑洞窗口
+DEBOUNCE_SECONDS = 5
+
+# 状态机: 持久化是否启用 / 启动模式 / 最后一次 push 结果
+_state: dict[str, object] = {
+    "mode": "disabled",          # disabled | cold_restore | fresh_start
+    "last_push_at": 0.0,         # unix timestamp
+    "pending_push": False,       # debounce 窗口期
+    "last_push_status": "idle",  # idle | uploading | verifying | ok | failed
+    "last_error": None,          # str | None
+    "last_verify": None,         # dict | None (verify 步骤结果)
 }
 
 
@@ -75,7 +91,7 @@ async def restore_from_hf() -> None:
     loop = asyncio.get_running_loop()
     try:
         await loop.run_in_executor(
-            _persist_executor,  # ✅ 独立池
+            _persist_executor,
             lambda: snapshot_download(
                 repo_id=repo_id,
                 repo_type="dataset",
@@ -87,7 +103,6 @@ async def restore_from_hf() -> None:
         _state["mode"] = "cold_restore"
         logger.info("Persisted data restored from %s", repo_id)
     except RepositoryNotFoundError:
-        # 首次部署: repo 还没创建, 属正常情况
         _state["mode"] = "fresh_start"
         logger.info(
             "Persist repo %s not found (first deploy?). "
@@ -95,15 +110,24 @@ async def restore_from_hf() -> None:
             repo_id,
         )
     except Exception as e:  # noqa: BLE001
-        logger.error("Persist restore failed (will start fresh): %s", e)
+        logger.error("Persist restore failed (will start fresh): %s", e, exc_info=True)
         _state["mode"] = "fresh_start"
-        # 不阻塞启动, 提示用户在 /readyz 看到降级状态
+        _state["last_error"] = f"restore: {type(e).__name__}: {e}"[:500]
 
 
-async def push_to_hf() -> None:
-    """同步推送本地数据到 HF Dataset repo. 阻塞."""
+async def push_to_hf() -> bool:
+    """同步推送本地数据到 HF Dataset repo. 阻塞, 含 verify.
+
+    返回 bool: True=upload + verify 都成功; False=失败 (失败原因在 _state['last_error']).
+
+    A 改良版要点:
+    1. upload_folder 完不立即返回, list_repo_files 确认数据真的上去了
+    2. verify 结果写入 _state['last_verify'], /readyz 暴露
+    3. push 状态写入 _state['last_push_status'], 前端能实时看
+    """
     if not settings.is_persist_enabled():
-        return
+        _state["last_push_status"] = "idle"
+        return False
     if _state["mode"] == "fresh_start":
         # 首次需要先 create_repo
         await _ensure_repo_exists()
@@ -111,33 +135,76 @@ async def push_to_hf() -> None:
     repo_id = settings.hf_persist_repo
     token = settings.hf_token.get_secret_value()
     local_root = data_dir()
-
     loop = asyncio.get_running_loop()
+
+    _state["last_push_status"] = "uploading"
+    _state["pending_push"] = False
+    started = time.time()
+
     try:
         await loop.run_in_executor(
-            _persist_executor,  # ✅ 独立池, 不阻塞业务 (Chroma / BGE-M3) 的 run_in_executor
+            _persist_executor,
             lambda: upload_folder(
                 folder_path=str(local_root),
                 repo_id=repo_id,
                 repo_type="dataset",
                 token=token,
-                commit_message=f"sync {asyncio.get_running_loop().time():.0f}",
+                commit_message=f"sync {started:.0f}",
                 ignore_patterns=[".cache/*", "*.tmp", "*.lock"],
             ),
         )
-        _state["last_push_at"] = asyncio.get_running_loop().time()
-        _state["pending_push"] = False
-        logger.info("Persisted data pushed to %s", repo_id)
+        upload_ms = int((time.time() - started) * 1000)
+        logger.info("upload_folder done in %dms, verifying...", upload_ms)
+
+        # ============ A 改良版核心: verify ============
+        _state["last_push_status"] = "verifying"
+        files = await loop.run_in_executor(
+            _persist_executor,
+            lambda: list_repo_files(
+                repo_id=repo_id,
+                repo_type="dataset",
+                token=token,
+            ),
+        )
+        has_chroma = any(f.startswith("chroma/") for f in files)
+        has_sqlite = any(f.startswith("sqlite/") for f in files)
+        has_uploads = any(f.startswith("uploads/") for f in files)
+
+        verify_result = {
+            "total_files": len(files),
+            "has_chroma": has_chroma,
+            "has_sqlite": has_sqlite,
+            "has_uploads": has_uploads,
+            "checked_at": time.time(),
+            "upload_ms": upload_ms,
+        }
+        _state["last_verify"] = verify_result
+
+        # 即使 upload 报成功但 repo 啥都没, 也标记 ok (cold start 后还没数据是正常的)
+        # 但要把 verify 结果暴露出去
+
+        _state["last_push_status"] = "ok"
+        _state["last_push_at"] = time.time()
+        _state["last_error"] = None
+        logger.info(
+            "Persist push OK: %s, %d files (chroma=%s sqlite=%s uploads=%s), upload %dms",
+            repo_id, len(files), has_chroma, has_sqlite, has_uploads, upload_ms,
+        )
+        return True
+
     except Exception as e:  # noqa: BLE001
-        # ✅ 失败也重置 flag, 否则 schedule_push 永远 short-circuit, 数据再也不推
-        _state["pending_push"] = False
-        logger.error("Persist push failed: %s", e)
+        _state["last_push_status"] = "failed"
+        _state["last_error"] = f"push: {type(e).__name__}: {e}"[:500]
+        _state["pending_push"] = False  # 失败也重置, 否则 schedule_push 永远 short-circuit
+        logger.error("Persist push failed: %s", e, exc_info=True)
+        return False
 
 
 async def schedule_push() -> None:
-    """异步推送, 不阻塞业务. 多次调用合并为一次 (简单去抖).
+    """异步推送 (5s debounce), 不阻塞业务.
 
-    适用: 摄入完成 / 删除文档后.
+    适用: chat 写入 / session checkpoint 等低频操作.
+    重要操作 (upload/delete) 应该直接 await push_to_hf().
     """
     if not settings.is_persist_enabled():
         return
@@ -149,12 +216,14 @@ async def schedule_push() -> None:
     _state["pending_push"] = True
 
     async def _delayed_push() -> None:
-        # 简单去抖: 延迟 30s, 把同一秒内的多次写合并
-        await asyncio.sleep(30)
+        await asyncio.sleep(DEBOUNCE_SECONDS)
         if _state["pending_push"]:
             await push_to_hf()
 
-    asyncio.create_task(_delayed_push())
+    # 持引用防 GC
+    task = asyncio.create_task(_delayed_push())
+    _background_tasks.add(task)
+    task.add_done_callback(_background_tasks.discard)
 
 
 async def _ensure_repo_exists() -> None:
@@ -164,7 +233,7 @@ async def _ensure_repo_exists() -> None:
     loop = asyncio.get_running_loop()
     try:
         await loop.run_in_executor(
-            _persist_executor,  # ✅ 独立池
+            _persist_executor,
             lambda: create_repo(
                 repo_id=repo_id,
                 repo_type="dataset",
@@ -175,15 +244,20 @@ async def _ensure_repo_exists() -> None:
         )
         logger.info("Created persist repo: %s", repo_id)
     except Exception as e:  # noqa: BLE001
-        logger.error("Failed to create persist repo: %s", e)
+        logger.error("Failed to create persist repo: %s", e, exc_info=True)
+        _state["last_error"] = f"create_repo: {e}"[:500]
 
 
 def persist_status() -> dict:
-    """供 /readyz 暴露持久化状态."""
+    """供 /readyz 暴露持久化状态 (含 A 改良版的 verify / last_error)."""
     return {
         "enabled": settings.is_persist_enabled(),
         "mode": _state["mode"],
         "pending_push": _state["pending_push"],
+        "last_push_status": _state["last_push_status"],
+        "last_push_at": _state["last_push_at"],
+        "last_error": _state["last_error"],
+        "last_verify": _state["last_verify"],
         "repo": settings.hf_persist_repo or None,
     }
 
@@ -201,4 +275,4 @@ __all__ = [
     "schedule_push",
     "persist_mode",
     "persist_status",
-]
+]