fix: 修复v6备份blob静默丢失 + 新增日志查询API key功能

blob丢失根因修复：
- store_blob推送Hub改为带3次重试+指数退避，成功置hub_ok标记
- commit前新增ensure_blobs_on_hub强制确认所有blob已落Hub，失败拒绝提交
- batch_check命中项增加可用性校验（本地+Hub清单），内容已丢失的降级为miss让客户端重传
- get_blob_local_path从Hub回源成功时标记hub_ok，Space重建后避免冗余重传
- backup_v6_blobs新增hub_ok列（幂等迁移），纯本地模式不受影响

日志查询API key（可吊销、随机生成）：
- 新表log_api_keys，admin端点：POST/GET /admin/api/log-keys、POST /admin/api/log-keys/revoke
- 新路由/api/logs：凭X-XTC-Log-Key头或?key=参数查询请求日志（列表+详情），
不在request_log记录范围内，key不落日志

.gitignore

@@ -35,3 +35,7 @@ ENV/
 # OS
 .DS_Store
 Thumbs.db
+
+_smoke_deps/
+_smoke_tmp/
+_commit_msg.txt

app/api/admin.py

@@ -184,6 +184,43 @@ async def revoke_token(
     return ok_with_cors({"revoked": ok})
 
 
+# ===== 日志查询 API Key（凭 key 调用 /api/logs，可吊销）=====
+
+@router.post("/log-keys")
+async def generate_log_key(
+    body: dict = Body(default={}),
+    _admin: str = Depends(_require_admin),
+):
+    """生成日志查询 API key。完整 key 仅本次响应返回一次。"""
+    from ..services import log_api_key_store
+
+    name = str(body.get("name") or "").strip()
+    info = log_api_key_store.generate_key(name)
+    return ok_with_cors(info)
+
+
+@router.get("/log-keys")
+async def list_log_keys(_admin: str = Depends(_require_admin)):
+    from ..services import log_api_key_store
+
+    return ok_with_cors({"keys": log_api_key_store.list_keys()})
+
+
+@router.post("/log-keys/revoke")
+async def revoke_log_key(
+    body: dict = Body(default={}),
+    _admin: str = Depends(_require_admin),
+):
+    """吊销日志查询 API key（body: {key: 完整 key}）。"""
+    from ..services import log_api_key_store
+
+    key = body.get("key")
+    if not key:
+        raise HttpError("key is required", status=400, code="bad_request")
+    revoked = log_api_key_store.revoke_key(str(key))
+    return ok_with_cors({"revoked": revoked})
+
+
 # ===== 配置 =====
 
 @router.get("/config")

app/api/log_query.py

@@ -0,0 +1,72 @@
+"""日志查询 API（凭 log API key 鉴权）：/api/logs/*
+
+用途：不依赖 admin key，凭可吊销的随机 API key 查询请求日志，
+方便外部工具/脚本排查线上问题（key 由 /admin/api/log-keys 生成）。
+
+鉴权方式（二选一）：
+- 请求头 X-XTC-Log-Key: <key>（推荐，避免 key 进代理/访问日志）
+- query 参数 ?key=<key>
+
+注意：/api/logs 不在 request_log 中间件的记录范围内（只记录 /v1/ /admin/api/ /u/api/），
+查询行为本身不会被记录，避免 key 泄露到日志存储。
+"""
+from __future__ import annotations
+
+from typing import Optional
+
+from fastapi import APIRouter, Query, Request
+
+from ..errors import HttpError
+from ..services import log_api_key_store, request_log_store
+from ._common import ok_with_cors
+
+router = APIRouter(prefix="/api/logs", tags=["log-query"])
+
+
+def _require_log_key(request: Request) -> str:
+    key = request.headers.get("x-xtc-log-key") or request.query_params.get("key") or ""
+    if not key:
+        raise HttpError(
+            "missing log api key: pass header X-XTC-Log-Key or query ?key=",
+            status=401, code="unauthorized",
+        )
+    if not log_api_key_store.verify_key(key):
+        raise HttpError(
+            "log api key invalid or revoked",
+            status=401, code="unauthorized",
+        )
+    return key
+
+
+@router.get("")
+async def list_logs(
+    request: Request,
+    limit: int = Query(default=50, ge=1, le=500),
+    offset: int = Query(default=0, ge=0),
+    path: Optional[str] = Query(default=None),
+    method: Optional[str] = Query(default=None),
+    status: Optional[str] = Query(default=None),
+    ok: Optional[int] = Query(default=None, ge=0, le=1),
+    hours: Optional[int] = Query(default=None, ge=1, le=720),
+    keyword: Optional[str] = Query(default=None),
+    category: Optional[str] = Query(default="business"),
+    device_id: Optional[str] = Query(default=None),
+):
+    _require_log_key(request)
+    result = request_log_store.list_records(
+        limit=limit, offset=offset,
+        path=path, method=method, status=status,
+        ok=bool(ok) if ok is not None else None,
+        hours=hours, keyword=keyword, category=category,
+        device_id=device_id,
+    )
+    return ok_with_cors(result)
+
+
+@router.get("/{rid}")
+async def get_log_detail(rid: int, request: Request):
+    _require_log_key(request)
+    rec = request_log_store.get_record(rid)
+    if not rec:
+        raise HttpError(f"request log not found: {rid}", status=404, code="not_found")
+    return ok_with_cors({"record": rec})

app/api/user_backups_v6.py

@@ -148,6 +148,18 @@ async def commit_manifest(set_id: str, request: Request):
             "blob_missing", "manifest 引用了未上传的 blob",
             missing_keys=still_missing[:20],
         )
+    # ★ 提交前强制确认所有 blob 已落 Hub（堵住静默丢失：本地有但 Hub 没有，
+    # Space 重启清盘后 blob 永久丢失。失败则拒绝提交，客户端可重试）
+    try:
+        push_failed = await backup_v6_store.ensure_blobs_on_hub(keys)
+    except Exception as e:
+        raise _err("blob_push_failed", f"blob 同步到 Hub 失败: {e}", status=503)
+    if push_failed:
+        raise _err(
+            "blob_push_failed", "部分 blob 同步到 Hub 失败，请稍后重试提交",
+            push_failed=push_failed[:20],
+            status=503,
+        )
     try:
         result = backup_v6_store.commit_set(set_id, manifest, uid)
     except ValueError as e:
@@ -342,7 +354,12 @@ async def batch_check(request: Request):
         except Exception:
             mtime = 0
         cleaned.append({"path": path, "size": size, "mtime": mtime})
-    result = backup_v6_store.batch_check(uid, cleaned)
+    # batch_check 现含 blob 可用性校验（本地缺失时有一次 Hub 清单网络往返），
+    # 必须放 to_thread，否则阻塞事件循环
+    try:
+        result = await asyncio.to_thread(backup_v6_store.batch_check, uid, cleaned)
+    except Exception as e:
+        raise _err("batch_check_failed", str(e), status=500)
     return ok_with_cors(result)
 
 

app/database.py

@@ -365,10 +365,23 @@ def _create_schema(conn: sqlite3.Connection) -> None:
             device          TEXT
         );
         CREATE INDEX IF NOT EXISTS idx_backup_v6_sets_user ON backup_v6_sets(user_id, created_at DESC);
+
+        -- 日志查询 API Key（供外部凭 key 调用 /api/logs 查询请求日志，可吊销）
+        CREATE TABLE IF NOT EXISTS log_api_keys (
+            key          TEXT PRIMARY KEY,
+            name         TEXT,
+            created_at   INTEGER NOT NULL,
+            revoked_at   INTEGER NOT NULL DEFAULT 0,
+            last_used_at INTEGER NOT NULL DEFAULT 0
+        );
         """
     )
     # 兼容已存在的库：若旧表无 device_id 列则补加（ALTER TABLE 幂等检查）
     _ensure_column(conn, "request_log", "device_id", "TEXT")
+    # v6 备份 blob 增加 hub_ok 标记：1 = 已确认推送到 HF Hub。
+    # 修复 blob 静默丢失：此前 fire-and-forget 推送失败无感知，Space 重启后
+    # 本地盘被清空导致 blob 永久丢失（manifest 还在但内容没了）。
+    _ensure_column(conn, "backup_v6_blobs", "hub_ok", "INTEGER NOT NULL DEFAULT 0")
     # file_meta 增加 alias 列：原始文件名（可能是中文），filename 改存随机英文名
     _ensure_column(conn, "file_meta", "alias", "TEXT")
     # user_tokens 增加 revoked_at 列：token rotation 宽限期支持

app/main.py

@@ -261,6 +261,7 @@ def create_app() -> FastAPI:
         admin_data,
         health,
         image_fix,
+        log_query,
         logs,
         music,
         music_records,
@@ -292,6 +293,7 @@ def create_app() -> FastAPI:
     app.include_router(usage_audit.router)
     app.include_router(webhooks.router)
     app.include_router(request_logs.router)
+    app.include_router(log_query.router)
     app.include_router(admin_html_router)
     app.include_router(user_html_router)
     app.include_router(user_files.router)

app/services/backup_v6_store.py

@@ -120,6 +120,15 @@ def store_blob(content: bytes, mime: str) -> dict:
                 try:
                     p.parent.mkdir(parents=True, exist_ok=True)
                     p.write_bytes(content)
+                    # 本地曾丢失 → hub_ok 状态不可信，重置为 0，
+                    # 由后台推送 / commit 前的 ensure_blobs_on_hub 重新确认
+                    try:
+                        with get_conn() as conn:
+                            conn.execute(
+                                "UPDATE backup_v6_blobs SET hub_ok = 0 WHERE key = ?", (key,)
+                            )
+                    except Exception:
+                        pass
                     if is_hub_enabled():
                         asyncio.ensure_future(_push_blob_to_hub(key))
                 except Exception as e:
@@ -142,16 +151,111 @@ def store_blob(content: bytes, mime: str) -> dict:
     return {"key": key, "size": size, "exists": False}
 
 
-async def _push_blob_to_hub(key: str) -> None:
+async def _push_blob_to_hub(key: str) -> bool:
+    """推送 blob 到 Hub（带重试 + 指数退避）。成功返回 True 并置 hub_ok=1。
+
+    v6 实际运行踩坑：fire-and-forget 单次推送失败即静默丢失，Space 重启后
+    本地盘清空 → blob 永久丢失（manifest 还在，内容没了，恢复 404）。
+    """
     p = _blob_local_path(key)
     if not p.exists():
-        return
+        return False
+    sub_key = f"{key[:2]}/{key}"
     try:
-        sub_key = f"{key[:2]}/{key}"
-        upload_bytes(NS_BLOBS, sub_key, p.read_bytes())
-        await push_to_hub(NS_BLOBS, sub_key)
+        data = p.read_bytes()
     except Exception as e:
-        logger.warning("[v6_blob] push to hub failed key=%s: %s", key[:8], e)
+        logger.warning("[v6_blob] read local file failed key=%s: %s", key[:8], e)
+        return False
+    max_retries = 3
+    for attempt in range(max_retries):
+        try:
+            upload_bytes(NS_BLOBS, sub_key, data)
+            ok = await push_to_hub(NS_BLOBS, sub_key)
+            if ok:
+                try:
+                    with get_conn() as conn:
+                        conn.execute(
+                            "UPDATE backup_v6_blobs SET hub_ok = 1 WHERE key = ?", (key,)
+                        )
+                except Exception as e:
+                    logger.warning("[v6_blob] mark hub_ok failed key=%s: %s", key[:8], e)
+                return True
+            # push_to_hub 内部已捕获异常返回 False，走重试
+            logger.warning(
+                "[v6_blob] push to hub returned false (attempt %d/%d) key=%s",
+                attempt + 1, max_retries, key[:8],
+            )
+        except Exception as e:
+            logger.warning(
+                "[v6_blob] push to hub failed (attempt %d/%d) key=%s: %s",
+                attempt + 1, max_retries, key[:8], e,
+            )
+        if attempt < max_retries - 1:
+            await asyncio.sleep(2 ** attempt)  # 1s / 2s 指数退避
+    logger.error("[v6_blob] push to hub FAILED after %d retries key=%s", max_retries, key[:8])
+    return False
+
+
+async def ensure_blobs_on_hub(keys: list[str], concurrency: int = 4) -> list[str]:
+    """确保 blob 已落到 Hub（按 hub_ok 标记）。
+
+    commit 前调用，堵住"manifest 提交成功但 blob 没上 Hub"的静默丢失漏洞：
+    - hub_ok=1 的跳过
+    - 本地有文件：并发推送（带重试），成功置 hub_ok=1
+    - 本地没文件：查 Hub 清单确认，已在 Hub 的直接置 hub_ok=1（Space 重建场景，
+      避免对已在 Hub 的 blob 做冗余重传）
+    返回仍失败的 key 列表（调用方应拒绝提交）。
+    """
+    uniq = [k for k in dict.fromkeys(keys) if k]
+    if not uniq:
+        return []
+    # 纯本地模式（未配置 HF_TOKEN / HF_CONFIG_REPO）：降级为不做 Hub 校验，
+    # 否则所有 blob 都会被误判为推送失败、拒绝提交
+    if not is_hub_enabled():
+        return []
+    placeholders = ",".join("?" * len(uniq))
+    with get_conn() as conn:
+        rows = conn.execute(
+            f"SELECT key, hub_ok FROM backup_v6_blobs WHERE key IN ({placeholders})", uniq
+        ).fetchall()
+    need = [str(r["key"]) for r in rows if not int(r["hub_ok"] or 0)]
+    if not need:
+        return []
+
+    have_local = [k for k in need if _blob_local_path(k).exists()]
+    no_local = [k for k in need if k not in set(have_local)]
+
+    confirmed: list[str] = []
+    if no_local and is_hub_enabled():
+        try:
+            hub_set = set(await asyncio.to_thread(list_hub_keys, NS_BLOBS))
+            confirmed = [k for k in no_local if k in hub_set]
+        except Exception as e:
+            logger.warning("[v6_blob] list hub keys failed: %s", e)
+    if confirmed:
+        try:
+            with get_conn() as conn:
+                for k in confirmed:
+                    conn.execute(
+                        "UPDATE backup_v6_blobs SET hub_ok = 1 WHERE key = ?", (k,)
+                    )
+        except Exception as e:
+            logger.warning("[v6_blob] mark confirmed hub_ok failed: %s", e)
+
+    failed: list[str] = []
+    if have_local:
+        sem = asyncio.Semaphore(max(1, concurrency))
+
+        async def _one(k: str) -> None:
+            async with sem:
+                ok = await _push_blob_to_hub(k)
+                if not ok:
+                    failed.append(k)
+
+        await asyncio.gather(*[_one(k) for k in have_local])
+    # 本地与 Hub 都没有的也返回（调用方按 blob_missing 处理）
+    lost = [k for k in no_local if k not in set(confirmed)]
+    return failed + lost
 
 
 def get_blob_local_path(key: str) -> Optional[Path]:
@@ -165,6 +269,14 @@ def get_blob_local_path(key: str) -> Optional[Path]:
             try:
                 p.parent.mkdir(parents=True, exist_ok=True)
                 p.write_bytes(data)
+                # 从 Hub 成功拉回 = Hub 上确实有，标记 hub_ok
+                try:
+                    with get_conn() as conn:
+                        conn.execute(
+                            "UPDATE backup_v6_blobs SET hub_ok = 1 WHERE key = ?", (key,)
+                        )
+                except Exception:
+                    pass
                 return p
             except Exception as e:
                 logger.warning("[v6_blob] fetch from hub failed key=%s: %s", key[:8], e)
@@ -640,9 +752,48 @@ def batch_check(user_id: str, paths: list[dict]) -> dict:
             hits.append({"path": path, "key": prev["key"], "size": size, "mtime": mtime})
         else:
             misses.append({"path": path})
+
+    # ★ 可用性过滤：命中但 blob 内容已丢失（本地 + Hub 都没有）的降级为 miss，
+    # 让客户端重传。否则客户端跳过这些文件 → commit 时 blob_missing。
+    # （v6 实际故障：Hub 推送静默失败 + Space 重启清盘 → manifest 还在、内容没了）
+    if hits:
+        lost_keys = set(_unavailable_blob_keys([h["key"] for h in hits]))
+        if lost_keys:
+            kept = []
+            for h in hits:
+                if h["key"] in lost_keys:
+                    logger.warning(
+                        "[v6_batch_check] blob content lost, downgrade to miss "
+                        "key=%s path=%s", h["key"][:8], h["path"],
+                    )
+                    misses.append({"path": h["path"]})
+                else:
+                    kept.append(h)
+            hits = kept
     return {"hits": hits, "misses": misses}
 
 
+def _unavailable_blob_keys(keys: list[str]) -> list[str]:
+    """返回本地与 Hub 都没有内容的 key 列表（含一次 Hub 清单往返，调用方放 to_thread）。
+
+    Hub 清单列举失败时返回空列表（宁可信其有，避免误判引发全量重传）。
+    """
+    uniq = [k for k in dict.fromkeys(keys) if k]
+    if not uniq:
+        return []
+    missing_local = [k for k in uniq if not _blob_local_path(k).exists()]
+    if not missing_local:
+        return []
+    if not is_hub_enabled():
+        return missing_local
+    try:
+        hub_keys = set(list_hub_keys(NS_BLOBS))
+    except Exception as e:
+        logger.warning("[v6_batch_check] list hub keys failed: %s", e)
+        return []
+    return [k for k in missing_local if k not in hub_keys]
+
+
 # ===== ZIP 分卷打包 =====
 
 def _volume_count(total_size: int) -> int:

app/services/log_api_key_store.py

@@ -0,0 +1,92 @@
+"""日志查询 API Key 管理。
+
+用途：生成可吊销的随机 API Key，凭 key 调用 /api/logs 查询请求日志
+（不需要 admin key，方便外部工具/脚本排查线上问题）。
+
+- key 随机生成（xtclog_ 前缀 + 32 字节 urlsafe 随机数），只在建时完整返回一次
+- 可随时吊销（revoked_at > 0 即失效，立即生效）
+- 记录 last_used_at 便于审计
+"""
+from __future__ import annotations
+
+import secrets
+import time
+from typing import Optional
+
+from ..database import get_conn
+
+KEY_PREFIX = "xtclog_"
+
+
+def _now() -> int:
+    return int(time.time())
+
+
+def generate_key(name: str = "") -> dict:
+    """生成新 key。返回 {key, name, createdAt}（key 仅此一次完整返回）。"""
+    key = KEY_PREFIX + secrets.token_urlsafe(24)
+    now = _now()
+    with get_conn() as conn:
+        conn.execute(
+            "INSERT INTO log_api_keys(key, name, created_at, revoked_at, last_used_at) "
+            "VALUES(?, ?, ?, 0, 0)",
+            (key, str(name or "")[:100], now),
+        )
+    return {"key": key, "name": str(name or "")[:100], "createdAt": now}
+
+
+def list_keys() -> list[dict]:
+    """列出全部 key（脱敏：只显示前缀 + 前 6 位）。"""
+    with get_conn() as conn:
+        rows = conn.execute(
+            "SELECT key, name, created_at, revoked_at, last_used_at "
+            "FROM log_api_keys ORDER BY created_at DESC"
+        ).fetchall()
+    out = []
+    for r in rows:
+        k = str(r["key"])
+        out.append({
+            "keyMasked": k[:len(KEY_PREFIX) + 6] + "..." if len(k) > len(KEY_PREFIX) + 6 else k,
+            "name": r["name"] or "",
+            "createdAt": int(r["created_at"]),
+            "revoked": int(r["revoked_at"] or 0) > 0,
+            "revokedAt": int(r["revoked_at"]) if int(r["revoked_at"] or 0) > 0 else None,
+            "lastUsedAt": int(r["last_used_at"] or 0) or None,
+        })
+    return out
+
+
+def revoke_key(key: str) -> bool:
+    """吊销 key（支持完整 key 或脱敏前缀匹配的完整 key）。返回是否吊销成功。"""
+    k = str(key or "").strip()
+    if not k:
+        return False
+    now = _now()
+    with get_conn() as conn:
+        cur = conn.execute(
+            "UPDATE log_api_keys SET revoked_at = ? "
+            "WHERE key = ? AND revoked_at = 0",
+            (now, k),
+        )
+        return cur.rowcount > 0
+
+
+def verify_key(key: str) -> bool:
+    """校验 key 有效（存在且未吊销），成功时更新 last_used_at。"""
+    k = str(key or "").strip()
+    if not k.startswith(KEY_PREFIX):
+        return False
+    now = _now()
+    with get_conn() as conn:
+        row = conn.execute(
+            "SELECT revoked_at FROM log_api_keys WHERE key = ?", (k,)
+        ).fetchone()
+        if not row or int(row["revoked_at"] or 0) > 0:
+            return False
+        try:
+            conn.execute(
+                "UPDATE log_api_keys SET last_used_at = ? WHERE key = ?", (now, k)
+            )
+        except Exception:
+            pass
+    return True