sync from GitHub 1c1ef96: feat: implement v6 backup storage layer with content-addressable blobs, snapshot management, and garbage collection

app/api/user_backups_v6.py

@@ -0,0 +1,393 @@
+"""v6 备份 API：/u/api/backups/v6/*
+
+端点（12 个）：
+- GET    /sets                           列出当前用户备份点
+- POST   /sets                           创建待提交备份点
+- GET    /sets/{setId}                   获取备份点 manifest
+- DELETE /sets/{setId}                   删除备份点（快照模型：任意点可删）
+- POST   /sets/{setId}/manifest          提交 manifest（原子生效）
+- POST   /sets/{setId}/rename            备份点改名
+- GET    /sets/{setId}/archive/info      ZIP 分卷信息
+- GET    /sets/{setId}/archive?volume=N  下载第 N 卷 ZIP（request.download 通道）
+- POST   /blobs/upload                   上传单个 blob（multipart，小文件）
+- POST   /blobs/upload_part              大文件 base64 分片上传
+- POST   /blobs/batch_check              卸载重装兜底（path+size+mtime 比对最新备份点）
+- DELETE /legacy-v5                      清理当前用户的 v5 遗留数据（被污染，整体废弃）
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from fastapi import APIRouter, File, Form, Request, UploadFile
+from fastapi.responses import FileResponse, StreamingResponse
+from typing import Optional
+
+from ..errors import HttpError
+from ..services import backup_v6_store
+from ._common import CORS_HEADERS, ok_with_cors, read_json_body
+from .user_files import require_user, require_user_for_download
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/u/api/backups/v6", tags=["user-backups-v6"])
+
+# 单 blob 上限：框架单文件硬限制 10MB，但恢复走 ZIP 分卷（每卷预算 8MB），
+# 单 blob 必须给卷内 manifest.json 留余量，与前端 backup-v6.js 的 MAX_BLOB_SIZE 一致
+MAX_BLOB_SIZE = 7 * 1024 * 1024
+
+# 后台 Hub 同步任务引用（保存强引用防 task 被 GC 中途回收）
+_sync_task_ref = None
+
+
+def _err(code: str, message: str, status: int = 400, **extra) -> HttpError:
+    return HttpError(message, status=status, code=code, details=extra or None)
+
+
+def _user_id(request: Request) -> str:
+    payload = require_user(request)
+    uid = str(payload.get("sub") or "")
+    if not uid:
+        raise _err("unauthorized", "missing user", status=401)
+    return uid
+
+
+# ===== sets =====
+
+@router.get("/sets")
+async def list_sets(request: Request):
+    uid = _user_id(request)
+    try:
+        limit = int(request.query_params.get("limit") or "30")
+        offset = int(request.query_params.get("offset") or "0")
+    except Exception:
+        limit, offset = 30, 0
+    # 本地 DB 为空时触发 Hub 同步（HF Space 重建后 DB 丢失）
+    if backup_v6_store.count_sets(uid) == 0:
+        global _sync_task_ref
+        if _sync_task_ref is None or _sync_task_ref.done():
+            _sync_task_ref = asyncio.create_task(backup_v6_store._sync_sets_background())
+    items = backup_v6_store.list_sets(uid, limit=limit, offset=offset)
+    return ok_with_cors({"items": items, "count": len(items), "total": len(items)})
+
+
+@router.post("/sets")
+async def create_set(request: Request):
+    uid = _user_id(request)
+    body = await read_json_body(request)
+    alias = str(body.get("alias") or "").strip()[:200]
+    try:
+        result = backup_v6_store.create_pending_set(uid, alias=alias)
+    except ValueError as e:
+        msg = str(e)
+        if msg == "quota_exceeded":
+            raise _err("quota_exceeded", f"已达备份点上限（{backup_v6_store.MAX_SETS_PER_USER} 份），请先删除旧的备份", status=409)
+        raise _err("create_failed", msg, status=500)
+    # 小天才 fetch.fetch 对非 200 的 2xx（如 201）会触发 fail 回调，统一返回 200
+    return ok_with_cors(result, status=200)
+
+
+@router.get("/sets/{set_id}")
+async def get_set(set_id: str, request: Request):
+    uid = _user_id(request)
+    owner = backup_v6_store.get_set_owner(set_id)
+    if not owner:
+        raise _err("set_not_found", "备份点不存在", status=404)
+    if owner != uid:
+        raise _err("forbidden", "无权访问他人备份", status=403)
+    manifest = backup_v6_store.get_set_manifest(set_id)
+    if manifest is None:
+        raise _err("manifest_missing", "manifest 缺失", status=500)
+    return ok_with_cors({"manifest": manifest})
+
+
+@router.delete("/sets/{set_id}")
+async def delete_set(set_id: str, request: Request):
+    uid = _user_id(request)
+    owner = backup_v6_store.get_set_owner(set_id)
+    if not owner:
+        raise _err("set_not_found", "备份点不存在", status=404)
+    if owner != uid:
+        raise _err("forbidden", "无权删除他人备份", status=403)
+    try:
+        result = backup_v6_store.delete_set(set_id, uid)
+    except ValueError as e:
+        msg = str(e)
+        if msg == "set_not_found":
+            raise _err("set_not_found", "备份点不存在", status=404)
+        if msg == "forbidden":
+            raise _err("forbidden", "无权删除他人备份", status=403)
+        raise _err("delete_failed", msg, status=500)
+    return ok_with_cors({"deleted": True, "setId": set_id, **result})
+
+
+@router.post("/sets/{set_id}/manifest")
+async def commit_manifest(set_id: str, request: Request):
+    uid = _user_id(request)
+    body = await read_json_body(request)
+    manifest = body.get("manifest")
+    if not isinstance(manifest, dict):
+        raise _err("invalid_manifest", "manifest 必须是 JSON 对象")
+    if str(manifest.get("setId") or "") != set_id:
+        manifest["setId"] = set_id
+    blobs = manifest.get("blobs")
+    if not isinstance(blobs, list) or not blobs:
+        raise _err("invalid_manifest_blobs", "manifest.blobs 必须是非空数组")
+    keys = [str(b.get("key") or "") for b in blobs if isinstance(b, dict)]
+    if not keys:
+        raise _err("empty_blobs", "manifest.blobs 不能为空")
+    # blob 可用性校验（含 Hub 回源，网络往返放工作线程避免阻塞事件循环）
+    try:
+        still_missing = await asyncio.to_thread(backup_v6_store.validate_blob_keys, keys)
+    except Exception as e:
+        raise _err("validate_failed", str(e), status=500)
+    if still_missing:
+        # 本地与 Hub 都没有：清理孤儿 DB 记录，让客户端重传
+        await asyncio.to_thread(backup_v6_store.cleanup_missing_blob_records, still_missing)
+        raise _err(
+            "blob_missing", "manifest 引用了未上传的 blob",
+            missing_keys=still_missing[:20],
+        )
+    try:
+        result = backup_v6_store.commit_set(set_id, manifest, uid)
+    except ValueError as e:
+        msg = str(e)
+        if msg == "set_not_pending":
+            raise _err("set_not_pending", "备份点不在待提交状态（已提交或已过期）")
+        if msg == "invalid_manifest":
+            raise _err("invalid_manifest", "manifest 格式无效")
+        if msg == "invalid_manifest_version":
+            raise _err("invalid_manifest_version", "manifest.version 必须为 6")
+        if msg in ("invalid_manifest_blobs", "empty_blobs"):
+            raise _err(msg, "manifest.blobs 必须是非空数组")
+        if msg.startswith("blob_missing:"):
+            missing = msg.split(":", 1)[1].split(",")
+            raise _err("blob_missing", "manifest 引用了未上传的 blob", missing_keys=missing)
+        if msg == "quota_exceeded":
+            raise _err("quota_exceeded", "已达备份点上限", status=409)
+        raise _err("commit_failed", msg, status=500)
+    return ok_with_cors(result)
+
+
+@router.post("/sets/{set_id}/rename")
+async def rename_set(set_id: str, request: Request):
+    uid = _user_id(request)
+    body = await read_json_body(request)
+    alias = str(body.get("alias") or "").strip()[:200]
+    if not alias:
+        raise _err("invalid_alias", "别名不能为空")
+    try:
+        result = backup_v6_store.rename_set(set_id, uid, alias)
+    except ValueError as e:
+        msg = str(e)
+        if msg == "set_not_found":
+            raise _err("set_not_found", "备份点不存在", status=404)
+        if msg == "forbidden":
+            raise _err("forbidden", "无权操作他人备份", status=403)
+        raise _err("rename_failed", msg, status=500)
+    return ok_with_cors(result)
+
+
+# ===== ZIP 分卷下载 =====
+
+@router.get("/sets/{set_id}/archive/info")
+async def archive_info(set_id: str, request: Request):
+    uid = _user_id(request)
+    owner = backup_v6_store.get_set_owner(set_id)
+    if not owner:
+        raise _err("set_not_found", "备份点不存在", status=404)
+    if owner != uid:
+        raise _err("forbidden", "无权访问他人备份", status=403)
+    try:
+        info = backup_v6_store.archive_info(set_id)
+    except ValueError as e:
+        if str(e) == "set_not_found":
+            raise _err("set_not_found", "备份点不存在", status=404)
+        raise _err("archive_info_failed", str(e), status=500)
+    return ok_with_cors(info)
+
+
+@router.get("/sets/{set_id}/archive")
+async def download_archive(set_id: str, request: Request):
+    payload = require_user_for_download(request)
+    uid = str(payload.get("sub") or "")
+    if not uid:
+        raise _err("unauthorized", "missing user", status=401)
+    owner = backup_v6_store.get_set_owner(set_id)
+    if not owner:
+        raise _err("set_not_found", "备份点不存在", status=404)
+    if owner != uid:
+        raise _err("forbidden", "无权下载他人备份", status=403)
+    try:
+        volume = int(request.query_params.get("volume") or "0")
+    except Exception:
+        volume = 0
+    try:
+        zip_path, zip_size = await asyncio.to_thread(
+            backup_v6_store.build_volume_zip, set_id, volume
+        )
+    except ValueError as e:
+        msg = str(e)
+        if msg == "set_not_found":
+            raise _err("set_not_found", "备份点不存在", status=404)
+        if msg == "volume_out_of_range":
+            raise _err("volume_out_of_range", "分卷号超出范围", status=400)
+        raise _err("zip_build_failed", msg, status=500)
+
+    filename = f"backup_{set_id[:8]}_v{volume}.zip"
+    generator = backup_v6_store.zip_iter_chunks(zip_path)
+    headers = {
+        "Content-Disposition": f'attachment; filename="{filename}"',
+        "Content-Length": str(zip_size),
+        **CORS_HEADERS,
+    }
+    return StreamingResponse(generator, media_type="application/zip", headers=headers)
+
+
+# ===== blobs =====
+
+@router.post("/blobs/upload")
+async def upload_blob(
+    request: Request,
+    file: UploadFile = File(...),
+    path: str = Form(""),
+    type: str = Form(""),
+):
+    uid = _user_id(request)
+    content = await file.read()
+    if len(content) > MAX_BLOB_SIZE:
+        raise _err("blob_too_large", f"blob 超过 {MAX_BLOB_SIZE} 字节", status=413)
+    mime = str(file.content_type or "")
+    if not mime:
+        mime = backup_v6_store._guess_mime(path or file.filename or "")
+    try:
+        result = backup_v6_store.store_blob(content, mime)
+    except ValueError as e:
+        if str(e) == "blob_too_large":
+            raise _err("blob_too_large", "blob 过大", status=413)
+        raise _err("store_failed", str(e), status=500)
+    return ok_with_cors(result)
+
+
+@router.post("/blobs/upload_part")
+async def upload_blob_part(request: Request):
+    """大文件分片上传：{uploadId, seq, dataB64, last} → {key, size, exists}。
+
+    流程：POST /blobs/upload_part/start 拿 uploadId → 逐片 POST → last=true 收尾。
+    """
+    uid = _user_id(request)
+    body = await read_json_body(request)
+    upload_id = str(body.get("uploadId") or "")
+    seq = body.get("seq")
+    data_b64 = body.get("dataB64")
+    last = body.get("last") is True
+    if not upload_id:
+        raise _err("invalid_upload_id", "uploadId 不能为空")
+    try:
+        seq = int(seq)
+    except Exception:
+        raise _err("invalid_seq", "seq 必须是整数")
+    if not isinstance(data_b64, str) or not data_b64:
+        raise _err("invalid_data", "dataB64 不能为空")
+    try:
+        result = backup_v6_store.upload_part(upload_id, uid, seq, data_b64, last)
+    except ValueError as e:
+        msg = str(e)
+        if msg == "upload_not_found":
+            raise _err("upload_not_found", "上传会话不存在或已过期", status=404)
+        if msg == "upload_expired":
+            raise _err("upload_expired", "上传会话已过期，请重新开始")
+        if msg == "part_too_large":
+            raise _err("part_too_large", f"单分片解码后超过 {backup_v6_store.MAX_PART_SIZE} 字节", status=413)
+        if msg == "invalid_base64":
+            raise _err("invalid_base64", "dataB64 不是合法 base64")
+        raise _err("upload_failed", msg, status=500)
+    return ok_with_cors(result)
+
+
+@router.post("/blobs/upload_part/start")
+async def upload_blob_part_start(request: Request):
+    uid = _user_id(request)
+    body = await read_json_body(request)
+    path = str(body.get("path") or "")
+    type_ = str(body.get("type") or "")
+    if not path:
+        raise _err("invalid_path", "path 不能为空")
+    upload_id = backup_v6_store.start_upload(uid, path, type_)
+    return ok_with_cors({"uploadId": upload_id, "expiresIn": backup_v6_store.PENDING_SET_TTL_SEC})
+
+
+@router.post("/blobs/batch_check")
+async def batch_check(request: Request):
+    uid = _user_id(request)
+    body = await read_json_body(request)
+    paths = body.get("paths")
+    if not isinstance(paths, list):
+        raise _err("invalid_paths", "paths 必须是数组")
+    if len(paths) > backup_v6_store.BATCH_CHECK_MAX:
+        raise _err("too_many_paths", f"单次最多 {backup_v6_store.BATCH_CHECK_MAX} 个 path")
+    cleaned = []
+    for p in paths:
+        if not isinstance(p, dict):
+            continue
+        path = str(p.get("path") or "").strip()
+        if not path:
+            continue
+        try:
+            size = int(p.get("size") or 0)
+        except Exception:
+            size = 0
+        try:
+            mtime = int(p.get("mtime") or 0)
+        except Exception:
+            mtime = 0
+        cleaned.append({"path": path, "size": size, "mtime": mtime})
+    result = backup_v6_store.batch_check(uid, cleaned)
+    return ok_with_cors(result)
+
+
+@router.get("/blobs/{key}")
+async def download_blob(key: str, request: Request):
+    payload = require_user_for_download(request)
+    uid = str(payload.get("sub") or "")
+    if not uid:
+        raise _err("unauthorized", "missing user", status=401)
+    if not _is_sha256_hex(key):
+        raise _err("invalid_key", "key 必须是 64 位 sha256 hex", status=400)
+    blob_path = backup_v6_store.get_blob_local_path(key)
+    if not blob_path or not blob_path.exists():
+        raise _err("blob_not_found", "blob 不存在", status=404)
+    meta = backup_v6_store.get_blob_meta(key)
+    mime = (meta or {}).get("mime") or "application/octet-stream"
+    headers = {
+        "Content-Disposition": f'attachment; filename="{blob_path.name}"',
+        **CORS_HEADERS,
+    }
+    return FileResponse(str(blob_path), media_type=mime, headers=headers)
+
+
+# ===== v5 遗留数据清理 =====
+
+@router.delete("/legacy-v5")
+async def purge_legacy_v5(request: Request):
+    """清理当前用户的 v5 遗留数据（被污染 + 设计缺陷，已整体废弃）。
+
+    删除该用户全部 v5 备份集、refs=0 的 v5 blob（本地 + Hub）。
+    v6 数据不受任何影响。
+    """
+    uid = _user_id(request)
+    try:
+        result = await asyncio.to_thread(backup_v6_store.purge_legacy_v5, uid)
+    except Exception as e:
+        raise _err("purge_failed", str(e), status=500)
+    return ok_with_cors({"ok": True, "purged": result})
+
+
+def _is_sha256_hex(s: str) -> bool:
+    if len(s) != 64:
+        return False
+    try:
+        int(s, 16)
+        return True
+    except Exception:
+        return False

app/database.py

@@ -336,6 +336,35 @@ def _create_schema(conn: sqlite3.Connection) -> None:
             alias           TEXT
         );
         CREATE INDEX IF NOT EXISTS idx_backup_sets_user ON backup_sets(user_id, created_at DESC);
+
+        -- v6 备份：内容寻址 blob（全局共享，refs 引用计数）
+        -- 与 v5 分表：v5 数据已被污染，v6 独立命名空间便于整体清理旧数据
+        CREATE TABLE IF NOT EXISTS backup_v6_blobs (
+            key           TEXT PRIMARY KEY,
+            size          INTEGER NOT NULL,
+            mime          TEXT,
+            refs          INTEGER NOT NULL DEFAULT 0,
+            uploaded_at   INTEGER NOT NULL
+        );
+        CREATE INDEX IF NOT EXISTS idx_backup_v6_blobs_refs ON backup_v6_blobs(refs);
+
+        -- v6 备份：备份点（快照模型：每个 set 的 manifest 都列全量文件清单，
+        -- blob 跨 set 共享。删除任意 set 只减 refs，其他 set 完全不受影响）
+        CREATE TABLE IF NOT EXISTS backup_v6_sets (
+            set_id          TEXT PRIMARY KEY,
+            user_id         TEXT NOT NULL,
+            alias           TEXT,
+            manifest        TEXT NOT NULL,
+            manifest_sha256 TEXT NOT NULL,
+            total_size      INTEGER NOT NULL,
+            blob_count      INTEGER NOT NULL,
+            created_at      INTEGER NOT NULL,
+            is_latest       INTEGER NOT NULL DEFAULT 0,
+            prev_set_id     TEXT,
+            mode            TEXT,
+            device          TEXT
+        );
+        CREATE INDEX IF NOT EXISTS idx_backup_v6_sets_user ON backup_v6_sets(user_id, created_at DESC);
         """
     )
     # 兼容已存在的库：若旧表无 device_id 列则补加（ALTER TABLE 幂等检查）

app/main.py

@@ -150,6 +150,10 @@ async def lifespan(app: FastAPI):
     # v5 备份：从 Hub 同步 manifest 元数据回本地 DB（HF Space 重建后本地 DB 丢失，
     # 但 Hub 上 backup_sets_v5/<user_id>/<set_id>.json 仍存在，需恢复否则管理面板和列表为空）
     asyncio.create_task(backup_v5_store._sync_manifests_background())
+    # v6 备份：GC 任务（refs=0 且超宽限期的 blob）+ Hub 元数据同步（Space 重建恢复）
+    from .services import backup_v6_store
+    backup_v6_store.start_gc_task()
+    asyncio.create_task(backup_v6_store._sync_sets_background())
     # 启动用户文件 Hub 定时同步任务
     s = get_settings()
     if is_hub_enabled():
@@ -176,6 +180,9 @@ async def lifespan(app: FastAPI):
     # 停止 v5 备份 GC 任务
     from .services import backup_v5_store
     await backup_v5_store.stop_gc_task()
+    # 停止 v6 备份 GC 任务
+    from .services import backup_v6_store
+    await backup_v6_store.stop_gc_task()
     if sync_task is not None:
         await user_files_sync.stop_sync_task()
     await close_http_client()
@@ -266,6 +273,7 @@ def create_app() -> FastAPI:
         user_files,
         user_backups,
         user_backups_v5,
+        user_backups_v6,
         webhooks,
         xtc,
     )
@@ -289,6 +297,7 @@ def create_app() -> FastAPI:
     app.include_router(user_files.router)
     app.include_router(user_backups.router)
     app.include_router(user_backups_v5.router)
+    app.include_router(user_backups_v6.router)
     app.include_router(music.router)
     app.include_router(music_records.router)
     app.include_router(logs.router)

app/services/backup_v6_store.py

@@ -0,0 +1,1098 @@
+"""v6 备份存储层：内容寻址 blob + 快照备份点 + ZIP 分卷 + GC + v5 数据清理。
+
+与 v5 的核心差异（针对 v5 实际运行中暴露的顽固问题重新设计）：
+1. 快照模型：每个备份点的 manifest 都列全量文件清单，blob 跨备份点共享。
+   删除任意备份点（包括最初那份）只减引用计数，其他备份点完全不受影响，
+   任意备份点可独立恢复。v5 的 prevSetId 增量链导致的"删中间点链断裂"
+   问题从模型上消除。
+2. 分片上传通道：≥512KB 的大文件可走 base64 分片（fetch.fetch JSON），
+   规避 request.upload multipart 对大文件回调不稳定的平台缺陷。
+3. ZIP 分卷打包：单卷 ≤8MB（框架单文件 10MB 硬限制留余量），大数据量
+   备份不再因"超 10MB 无法 ZIP"而被迫走会耗尽下载管理器的逐 blob 通道。
+   ZIP 内保持扁平结构 blob_<idx>.dat（小天才 zip.decompress 对子目录
+   ZIP 解压不可靠，v5 实测结论）。
+4. blob 双重校验：commit 时校验 DB 记录 + 本地文件，本地丢失自动从 Hub
+   拉回（HF Space 重建场景）。
+5. 独立命名空间 backup_v6_blobs / backup_v6_sets：与被污染的 v5 数据
+   完全隔离，v5 数据可整体清理（purge_legacy_v5）。
+
+存储布局：
+- 本地：/data/hf_bucket/backup_v6_blobs/<key[:2]>/<key>
+        /data/hf_bucket/backup_v6_sets/<user_id>/<set_id>.json
+- Hub：  同名空间（冷备，HF Space 重建后恢复）
+- DB：   backup_v6_blobs / backup_v6_sets 表
+"""
+from __future__ import annotations
+
+import asyncio
+import base64
+import hashlib
+import json
+import logging
+import time
+import zipfile
+from pathlib import Path
+from typing import Optional
+
+from ..database import get_conn
+from ..hf_storage import (
+    delete_from_hub,
+    download_bytes,
+    is_hub_enabled,
+    list_hub_keys,
+    push_to_hub,
+    upload_bytes,
+)
+
+logger = logging.getLogger(__name__)
+
+# ===== 常量 =====
+
+NS_BLOBS = "backup_v6_blobs"
+NS_SETS = "backup_v6_sets"
+
+MAX_BLOB_SIZE = 10 * 1024 * 1024        # 单 blob ≤10MB（框架硬限制）
+MAX_PART_SIZE = 1024 * 1024             # 单分片解码后 ≤1MB
+VOLUME_BUDGET = 8 * 1024 * 1024         # ZIP 单卷内容预算 ≤8MB（10MB 限制留余量）
+MAX_SETS_PER_USER = 8                   # 单用户保留最近 8 个备份点
+PENDING_SET_TTL_SEC = 600               # 待提交 set / 分片上传 TTL：10 分钟
+GC_INTERVAL_SEC = 3600                  # GC 周期：1 小时
+GC_BLOB_GRACE_SEC = 24 * 3600           # refs=0 的 blob 保留 24h 再删（防误删进行中的上传）
+BATCH_CHECK_MAX = 300                   # batch_check 单次最大 path 数
+
+# 模块级内存状态（多 worker 场景各自独立，可接受：失败重传即可）
+_pending_sets: "dict[str, dict]" = {}            # set_id -> {user_id, alias, created_at, prev_set_id}
+_pending_uploads: "dict[str, dict]" = {}         # uploadId -> {user_id, parts, created_at, path, type}
+
+
+def _now() -> int:
+    return int(time.time())
+
+
+def _blob_local_path(key: str) -> Path:
+    from ..hf_storage import _local_path
+    return _local_path(NS_BLOBS, f"{key[:2]}/{key}")
+
+
+def _manifest_local_path(user_id: str, set_id: str) -> Path:
+    from ..hf_storage import _local_path
+    return _local_path(NS_SETS, f"{user_id}/{set_id}.json")
+
+
+def _gen_id() -> str:
+    import uuid
+    return uuid.uuid4().hex
+
+
+def _cleanup_expired() -> None:
+    """清理过期的待提交 set 与分片上传缓冲。"""
+    now = _now()
+    expired = [sid for sid, p in _pending_sets.items() if now - p["created_at"] > PENDING_SET_TTL_SEC]
+    for sid in expired:
+        _pending_sets.pop(sid, None)
+    expired2 = [uid for uid, p in _pending_uploads.items() if now - p["created_at"] > PENDING_SET_TTL_SEC]
+    for uid in expired2:
+        _pending_uploads.pop(uid, None)
+
+
+# ===== Blob 存储（内容寻址）=====
+
+def store_blob(content: bytes, mime: str) -> dict:
+    """存 blob：key = sha256(content)。已存在则去重。
+
+    返回 {key, size, exists}。不增 refs（refs 在 commit_set 时统一累加）。
+    """
+    if len(content) > MAX_BLOB_SIZE:
+        raise ValueError("blob_too_large")
+    key = hashlib.sha256(content).hexdigest()
+    size = len(content)
+    now = _now()
+    p = _blob_local_path(key)
+
+    with get_conn() as conn:
+        row = conn.execute(
+            "SELECT key FROM backup_v6_blobs WHERE key = ?", (key,)
+        ).fetchone()
+        if row:
+            # DB 有记录但本地文件丢失（HF Space 重建后常见）：补写本地，
+            # 否则 commit 校验会报 blob_missing（v5 踩过的坑）
+            if not p.exists():
+                try:
+                    p.parent.mkdir(parents=True, exist_ok=True)
+                    p.write_bytes(content)
+                    if is_hub_enabled():
+                        asyncio.ensure_future(_push_blob_to_hub(key))
+                except Exception as e:
+                    logger.warning("[v6_blob] restore local file failed key=%s: %s", key[:8], e)
+            return {"key": key, "size": size, "exists": True}
+
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_bytes(content)
+    try:
+        with get_conn() as conn:
+            conn.execute(
+                "INSERT OR IGNORE INTO backup_v6_blobs(key, size, mime, refs, uploaded_at) "
+                "VALUES(?, ?, ?, 0, ?)",
+                (key, size, mime, now),
+            )
+    except Exception as e:
+        logger.warning("[v6_blob] insert db failed key=%s: %s", key[:8], e)
+    if is_hub_enabled():
+        asyncio.ensure_future(_push_blob_to_hub(key))
+    return {"key": key, "size": size, "exists": False}
+
+
+async def _push_blob_to_hub(key: str) -> None:
+    p = _blob_local_path(key)
+    if not p.exists():
+        return
+    try:
+        sub_key = f"{key[:2]}/{key}"
+        upload_bytes(NS_BLOBS, sub_key, p.read_bytes())
+        await push_to_hub(NS_BLOBS, sub_key)
+    except Exception as e:
+        logger.warning("[v6_blob] push to hub failed key=%s: %s", key[:8], e)
+
+
+def get_blob_local_path(key: str) -> Optional[Path]:
+    """返回 blob 本地路径；本地缺失时从 Hub 拉回；都没有返回 None。"""
+    p = _blob_local_path(key)
+    if p.exists() and p.is_file():
+        return p
+    if is_hub_enabled():
+        data = download_bytes(NS_BLOBS, f"{key[:2]}/{key}")
+        if data is not None:
+            try:
+                p.parent.mkdir(parents=True, exist_ok=True)
+                p.write_bytes(data)
+                return p
+            except Exception as e:
+                logger.warning("[v6_blob] fetch from hub failed key=%s: %s", key[:8], e)
+    return None
+
+
+def get_blob_meta(key: str) -> Optional[dict]:
+    with get_conn() as conn:
+        row = conn.execute(
+            "SELECT key, size, mime, refs, uploaded_at FROM backup_v6_blobs WHERE key = ?",
+            (key,),
+        ).fetchone()
+    if not row:
+        return None
+    d = dict(row)
+    d["uploadedAt"] = d.pop("uploaded_at")
+    return d
+
+
+# ===== 分片上传（大文件通道）=====
+
+def start_upload(user_id: str, path: str, type_: str) -> str:
+    """创建分片上传会话，返回 uploadId。"""
+    _cleanup_expired()
+    upload_id = _gen_id()
+    _pending_uploads[upload_id] = {
+        "user_id": user_id,
+        "parts": {},
+        "created_at": _now(),
+        "path": str(path or "")[:256],
+        "type": str(type_ or "")[:32],
+    }
+    return upload_id
+
+
+def upload_part(upload_id: str, user_id: str, seq: int, data_b64: str, last: bool) -> dict:
+    """接收一个分片（base64）。last=true 时拼接所有分片存为 blob。
+
+    返回 {key, size, exists}（非 last 时返回 {accepted: true}）。
+    """
+    up = _pending_uploads.get(upload_id)
+    if not up or up["user_id"] != user_id:
+        raise ValueError("upload_not_found")
+    if _now() - up["created_at"] > PENDING_SET_TTL_SEC:
+        _pending_uploads.pop(upload_id, None)
+        raise ValueError("upload_expired")
+
+    try:
+        raw = base64.b64decode(str(data_b64 or ""), validate=False)
+    except Exception:
+        raise ValueError("invalid_base64")
+    if len(raw) > MAX_PART_SIZE:
+        raise ValueError("part_too_large")
+    up["parts"][int(seq)] = raw
+
+    if not last:
+        return {"accepted": True}
+
+    # 拼接所有分片（按 seq 排序）
+    content = b"".join(up["parts"][s] for s in sorted(up["parts"].keys()))
+    mime = _guess_mime(up["path"])
+    result = store_blob(content, mime)
+    _pending_uploads.pop(upload_id, None)
+    result["size"] = len(content)
+    return result
+
+
+def _guess_mime(path: str) -> str:
+    ext = (path or "").rsplit(".", 1)[-1].lower() if "." in (path or "") else ""
+    return {
+        "json": "application/json",
+        "txt": "text/plain",
+        "md": "text/markdown",
+        "png": "image/png",
+        "jpg": "image/jpeg",
+        "jpeg": "image/jpeg",
+        "gif": "image/gif",
+        "webp": "image/webp",
+        "mp3": "audio/mpeg",
+        "wav": "audio/wav",
+    }.get(ext, "application/octet-stream")
+
+
+# ===== 备份点（快照）管理 =====
+
+def count_sets(user_id: str) -> int:
+    with get_conn() as conn:
+        row = conn.execute(
+            "SELECT COUNT(*) AS c FROM backup_v6_sets WHERE user_id = ?", (user_id,)
+        ).fetchone()
+    return int(row["c"]) if row else 0
+
+
+def create_pending_set(user_id: str, alias: Optional[str] = None) -> dict:
+    """创建待提交备份点，返回 {setId, prevSetId, serverLatest, expiresIn}。"""
+    _cleanup_expired()
+    with get_conn() as conn:
+        cnt = conn.execute(
+            "SELECT COUNT(*) AS c FROM backup_v6_sets WHERE user_id = ?", (user_id,)
+        ).fetchone()
+        if cnt and int(cnt["c"]) >= MAX_SETS_PER_USER:
+            raise ValueError("quota_exceeded")
+        row = conn.execute(
+            "SELECT set_id, created_at, alias, device FROM backup_v6_sets "
+            "WHERE user_id = ? AND is_latest = 1 LIMIT 1",
+            (user_id,),
+        ).fetchone()
+    prev_set_id = str(row["set_id"]) if row else ""
+    server_latest = None
+    if row:
+        server_latest = {
+            "setId": prev_set_id,
+            "exportedAt": int(row["created_at"]),
+            "alias": row["alias"] or "",
+            "device": row["device"] or "",
+        }
+    set_id = _gen_id()
+    _pending_sets[set_id] = {
+        "user_id": user_id,
+        "alias": alias or "",
+        "created_at": _now(),
+        "prev_set_id": prev_set_id,
+    }
+    return {
+        "setId": set_id,
+        "prevSetId": prev_set_id,
+        "serverLatest": server_latest,
+        "expiresIn": PENDING_SET_TTL_SEC,
+    }
+
+
+def get_set_owner(set_id: str) -> Optional[str]:
+    with get_conn() as conn:
+        row = conn.execute(
+            "SELECT user_id FROM backup_v6_sets WHERE set_id = ?", (set_id,)
+        ).fetchone()
+    return str(row["user_id"]) if row else None
+
+
+def get_set_manifest(set_id: str) -> Optional[dict]:
+    with get_conn() as conn:
+        row = conn.execute(
+            "SELECT manifest FROM backup_v6_sets WHERE set_id = ?", (set_id,)
+        ).fetchone()
+    if not row:
+        return None
+    try:
+        return json.loads(row["manifest"])
+    except Exception as e:
+        logger.warning("[v6_set] parse manifest failed set=%s: %s", set_id[:8], e)
+        return None
+
+
+def list_sets(user_id: str, limit: int = 20, offset: int = 0) -> list[dict]:
+    with get_conn() as conn:
+        rows = conn.execute(
+            "SELECT set_id, manifest, total_size, blob_count, created_at, "
+            "is_latest, prev_set_id, alias, mode, device "
+            "FROM backup_v6_sets WHERE user_id = ? "
+            "ORDER BY created_at DESC LIMIT ? OFFSET ?",
+            (user_id, int(limit), int(offset)),
+        ).fetchall()
+    out = []
+    for r in rows:
+        try:
+            m = json.loads(r["manifest"])
+        except Exception:
+            m = {}
+        stats = m.get("stats", {}) if isinstance(m, dict) else {}
+        out.append({
+            "setId": r["set_id"],
+            "exportedAt": (m.get("exportedAt") if isinstance(m, dict) else None) or int(r["created_at"]),
+            "totalSize": int(r["total_size"]),
+            "blobCount": int(r["blob_count"]),
+            "isLatest": bool(r["is_latest"]),
+            "alias": r["alias"] or "",
+            "prevSetId": r["prev_set_id"] or "",
+            "mode": r["mode"] or "",
+            "device": r["device"] or "",
+            "stats": stats,
+            "volumes": _volume_count(int(r["total_size"])),
+        })
+    return out
+
+
+def rename_set(set_id: str, user_id: str, alias: str) -> dict:
+    with get_conn() as conn:
+        row = conn.execute(
+            "SELECT user_id FROM backup_v6_sets WHERE set_id = ?", (set_id,)
+        ).fetchone()
+        if not row:
+            raise ValueError("set_not_found")
+        if str(row["user_id"]) != user_id:
+            raise ValueError("forbidden")
+        conn.execute(
+            "UPDATE backup_v6_sets SET alias = ? WHERE set_id = ?",
+            (alias[:200], set_id),
+        )
+    # 同步 Hub 上的 manifest（alias 也写在 manifest 里）
+    manifest = get_set_manifest(set_id)
+    if manifest is not None:
+        manifest["alias"] = alias[:200]
+        _write_manifest_to_local(user_id, set_id, manifest)
+        if is_hub_enabled():
+            asyncio.ensure_future(_push_manifest_to_hub(user_id, set_id))
+    return {"setId": set_id, "alias": alias[:200]}
+
+
+def validate_blob_keys(keys: list) -> list:
+    """校验 blob 可用性：DB 记录存在 + 本地文件存在（缺失时从 Hub 拉回）。
+
+    返回仍缺失的 key 列表（本地与 Hub 都没有）。
+    ★ 含 Hub 网络往返（HF Space 重建后回源），调用方应放 asyncio.to_thread。
+    """
+    if not keys:
+        return []
+    uniq = list(dict.fromkeys(k for k in keys if k))
+    placeholders = ",".join("?" * len(uniq))
+    with get_conn() as conn:
+        rows = conn.execute(
+            f"SELECT key FROM backup_v6_blobs WHERE key IN ({placeholders})", uniq
+        ).fetchall()
+    existing = {r["key"] for r in rows}
+    missing_db = [k for k in uniq if k not in existing]
+    # 本地文件缺失 → 从 Hub 拉回（HF Space 重建场景）
+    still_missing = []
+    for k in uniq:
+        if k in missing_db:
+            still_missing.append(k)
+            continue
+        p = get_blob_local_path(k)
+        if not (p and p.exists()):
+            still_missing.append(k)
+    return still_missing
+
+
+def cleanup_missing_blob_records(keys: list) -> None:
+    """删除本地与 Hub 都缺失的 blob 的 DB 记录（孤儿记录清理）。"""
+    for k in keys:
+        try:
+            with get_conn() as conn:
+                conn.execute("DELETE FROM backup_v6_blobs WHERE key = ?", (k,))
+        except Exception as e:
+            logger.warning("[v6_commit] cleanup orphan record failed key=%s: %s", k[:8], e)
+
+
+def commit_set(set_id: str, manifest: dict, user_id: str) -> dict:
+    """提交 manifest：校验（调用方先 validate_blob_keys）→ 事务写表 → refs 累加
+    → 配额淘汰最旧 → 推 Hub。
+
+    ★ 主写序列（UPDATE is_latest → INSERT set → UPDATE refs）包在
+      BEGIN IMMEDIATE 事务里：isolation_level=None 下每语句即时提交，
+      拆开写会在崩溃窗口内造成 refs 少计 → GC 误删在用 blob。
+    """
+    pending = _pending_sets.get(set_id)
+    if not pending or pending["user_id"] != user_id:
+        raise ValueError("set_not_pending")
+    if not isinstance(manifest, dict):
+        raise ValueError("invalid_manifest")
+    if int(manifest.get("version") or 0) != 6:
+        raise ValueError("invalid_manifest_version")
+    blobs = manifest.get("blobs")
+    if not isinstance(blobs, list) or not blobs:
+        raise ValueError("invalid_manifest_blobs")
+
+    keys = [str(b.get("key") or "") for b in blobs if isinstance(b, dict)]
+    if not keys:
+        raise ValueError("empty_blobs")
+
+    # 校验：DB 记录存在（Hub 回源已由调用方 validate_blob_keys 完成）
+    placeholders = ",".join("?" * len(keys))
+    with get_conn() as conn:
+        rows = conn.execute(
+            f"SELECT key FROM backup_v6_blobs WHERE key IN ({placeholders})", keys
+        ).fetchall()
+    existing_keys = {r["key"] for r in rows}
+    missing = [k for k in dict.fromkeys(keys) if k not in existing_keys]
+    if missing:
+        raise ValueError("blob_missing:" + ",".join(missing[:20]))
+
+    # checksum
+    manifest_for_hash = {k: v for k, v in manifest.items() if k != "checksum"}
+    checksum = hashlib.sha256(
+        json.dumps(manifest_for_hash, sort_keys=True, ensure_ascii=False).encode("utf-8")
+    ).hexdigest()
+    manifest["checksum"] = checksum
+
+    total_size = sum(int(b.get("size") or 0) for b in blobs if isinstance(b, dict))
+    blob_count = len(blobs)
+    now = _now()
+    prev_set_id = manifest.get("prevSetId") or pending.get("prev_set_id") or ""
+    alias = (manifest.get("alias") or pending.get("alias") or "")[:200]
+    mode = str(manifest.get("mode") or "")[:32]
+    device = str(manifest.get("device") or "")[:128]
+
+    # 配额：>=上限 先删最旧（快照模型：删除不影响其他备份点）
+    with get_conn() as conn:
+        cnt_row = conn.execute(
+            "SELECT COUNT(*) AS c FROM backup_v6_sets WHERE user_id = ?", (user_id,)
+        ).fetchone()
+        cnt = int(cnt_row["c"]) if cnt_row else 0
+    if cnt >= MAX_SETS_PER_USER:
+        with get_conn() as conn:
+            oldest = conn.execute(
+                "SELECT set_id FROM backup_v6_sets WHERE user_id = ? ORDER BY created_at ASC LIMIT 1",
+                (user_id,),
+            ).fetchone()
+        if oldest:
+            delete_set(str(oldest["set_id"]), user_id)
+
+    uniq_keys = list(dict.fromkeys(keys))
+    # 事务：is_latest 切换 + INSERT + refs 累加 原子完成
+    with get_conn() as conn:
+        try:
+            conn.execute("BEGIN IMMEDIATE")
+            conn.execute(
+                "UPDATE backup_v6_sets SET is_latest = 0 WHERE user_id = ? AND is_latest = 1",
+                (user_id,),
+            )
+            conn.execute(
+                "INSERT INTO backup_v6_sets(set_id, user_id, alias, manifest, manifest_sha256, "
+                "total_size, blob_count, created_at, is_latest, prev_set_id, mode, device) "
+                "VALUES(?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?)",
+                (
+                    set_id, user_id, alias, json.dumps(manifest, ensure_ascii=False), checksum,
+                    total_size, blob_count, now, prev_set_id, mode, device,
+                ),
+            )
+            for k in uniq_keys:
+                conn.execute(
+                    "UPDATE backup_v6_blobs SET refs = refs + 1 WHERE key = ?", (k,)
+                )
+            conn.execute("COMMIT")
+        except Exception:
+            try:
+                conn.execute("ROLLBACK")
+            except Exception:
+                pass
+            raise
+
+    _pending_sets.pop(set_id, None)
+
+    _write_manifest_to_local(user_id, set_id, manifest)
+    if is_hub_enabled():
+        asyncio.ensure_future(_push_manifest_to_hub(user_id, set_id))
+
+    return {
+        "setId": set_id,
+        "manifestSha256": checksum,
+        "totalSize": total_size,
+        "blobCount": blob_count,
+        "isLatest": True,
+    }
+
+
+def delete_set(set_id: str, user_id: str) -> dict:
+    """删除备份点：减 refs；refs=0 的 blob 交给 GC（宽限期后清理）。
+
+    快照模型下删除任意备份点（含最初那份）都不影响其他备份点。
+    """
+    with get_conn() as conn:
+        row = conn.execute(
+            "SELECT user_id, manifest FROM backup_v6_sets WHERE set_id = ?", (set_id,)
+        ).fetchone()
+    if not row:
+        raise ValueError("set_not_found")
+    if str(row["user_id"]) != user_id:
+        raise ValueError("forbidden")
+    try:
+        manifest = json.loads(row["manifest"])
+        keys = [str(b.get("key") or "") for b in manifest.get("blobs", []) if isinstance(b, dict)]
+    except Exception:
+        keys = []
+
+    freed_blobs = 0
+    freed_bytes = 0
+    with get_conn() as conn:
+        for k in dict.fromkeys(keys):
+            conn.execute(
+                "UPDATE backup_v6_blobs SET refs = MAX(refs - 1, 0) WHERE key = ?", (k,)
+            )
+            r = conn.execute(
+                "SELECT refs, size FROM backup_v6_blobs WHERE key = ?", (k,)
+            ).fetchone()
+            if r and int(r["refs"]) <= 0:
+                freed_blobs += 1
+                freed_bytes += int(r["size"])
+        conn.execute("DELETE FROM backup_v6_sets WHERE set_id = ?", (set_id,))
+
+    # 若删的是 latest，把最新的升级为 latest（保持"最新"语义，batch_check 依赖它）
+    with get_conn() as conn:
+        latest = conn.execute(
+            "SELECT set_id FROM backup_v6_sets WHERE user_id = ? AND is_latest = 1 LIMIT 1",
+            (user_id,),
+        ).fetchone()
+        if not latest:
+            newest = conn.execute(
+                "SELECT set_id FROM backup_v6_sets WHERE user_id = ? ORDER BY created_at DESC LIMIT 1",
+                (user_id,),
+            ).fetchone()
+            if newest:
+                conn.execute(
+                    "UPDATE backup_v6_sets SET is_latest = 1 WHERE set_id = ?",
+                    (str(newest["set_id"]),),
+                )
+
+    # 删本地 + Hub 的 manifest
+    try:
+        mp = _manifest_local_path(user_id, set_id)
+        if mp.exists():
+            mp.unlink(missing_ok=True)
+        if is_hub_enabled():
+            delete_from_hub(NS_SETS, f"{user_id}/{set_id}.json")
+    except Exception as e:
+        logger.warning("[v6_set] delete manifest failed set=%s: %s", set_id[:8], e)
+
+    return {"freedBlobs": freed_blobs, "freedBytes": freed_bytes}
+
+
+def _write_manifest_to_local(user_id: str, set_id: str, manifest: dict) -> None:
+    p = _manifest_local_path(user_id, set_id)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
+
+
+async def _push_manifest_to_hub(user_id: str, set_id: str) -> None:
+    p = _manifest_local_path(user_id, set_id)
+    if not p.exists():
+        return
+    try:
+        sub_key = f"{user_id}/{set_id}.json"
+        upload_bytes(NS_SETS, sub_key, p.read_bytes())
+        await push_to_hub(NS_SETS, sub_key)
+    except Exception as e:
+        logger.warning("[v6_set] push manifest to hub failed set=%s: %s", set_id[:8], e)
+
+
+def batch_check(user_id: str, paths: list[dict]) -> dict:
+    """卸载重装兜底：与该用户最新备份点的 manifest 按 path+size+mtime 比对。
+
+    返回 {hits: [{path, key}], misses: [{path}]}。
+    注意：比对基准是服务端最新备份点（不需要客户端传 prevSetId）。
+    """
+    with get_conn() as conn:
+        row = conn.execute(
+            "SELECT manifest FROM backup_v6_sets WHERE user_id = ? AND is_latest = 1 LIMIT 1",
+            (user_id,),
+        ).fetchone()
+    if not row:
+        return {"hits": [], "misses": [{"path": str(p.get("path") or "")} for p in paths]}
+    try:
+        manifest = json.loads(row["manifest"])
+    except Exception:
+        manifest = {}
+
+    prev_map = {}
+    for b in manifest.get("blobs", []):
+        if isinstance(b, dict):
+            prev_map[str(b.get("path") or "")] = {
+                "key": str(b.get("key") or ""),
+                "size": int(b.get("size") or 0),
+                "mtime": int(b.get("mtime") or 0),
+            }
+
+    hits = []
+    misses = []
+    for p in paths:
+        path = str(p.get("path") or "")
+        size = int(p.get("size") or 0)
+        mtime = int(p.get("mtime") or 0)
+        prev = prev_map.get(path)
+        if prev and prev["key"] and prev["size"] == size and prev["mtime"] == mtime:
+            hits.append({"path": path, "key": prev["key"], "size": size, "mtime": mtime})
+        else:
+            misses.append({"path": path})
+    return {"hits": hits, "misses": misses}
+
+
+# ===== ZIP 分卷打包 =====
+
+def _volume_count(total_size: int) -> int:
+    """按 8MB 预算估算卷数（供列表展示）。"""
+    if total_size <= 0:
+        return 1
+    return max(1, (total_size + VOLUME_BUDGET - 1) // VOLUME_BUDGET)
+
+
+def archive_info(set_id: str) -> dict:
+    """返回备份点的分卷信息：{totalSize, volumes, volumeSize, entries}。
+
+    entries: 每卷的 {volume, startIdx, endIdx(不含), blobCount, contentSize}
+    """
+    manifest = get_set_manifest(set_id)
+    if not manifest:
+        raise ValueError("set_not_found")
+    # 保持原始数组顺序（分卷下标与 build_volume_zip 的 blob_<i>.dat 对齐），
+    # 仅对求和做类型防御（commit 时已校验过全部元素，正常数据无影响）
+    blobs = manifest.get("blobs", []) or []
+    total_size = sum(
+        int(b.get("size") or 0) if isinstance(b, dict) else 0 for b in blobs
+    )
+
+    volumes = []
+    cur = None
+    for i, b in enumerate(blobs):
+        size = int(b.get("size") or 0) if isinstance(b, dict) else 0
+        if cur is None or cur["contentSize"] + size > VOLUME_BUDGET:
+            if cur is not None:
+                volumes.append(cur)
+            cur = {"volume": len(volumes), "startIdx": i, "endIdx": i + 1,
+                   "blobCount": 1, "contentSize": size}
+        else:
+            cur["endIdx"] = i + 1
+            cur["blobCount"] += 1
+            cur["contentSize"] += size
+    if cur is not None:
+        volumes.append(cur)
+    if not volumes:
+        volumes = [{"volume": 0, "startIdx": 0, "endIdx": 0, "blobCount": 0, "contentSize": 0}]
+
+    return {
+        "setId": set_id,
+        "totalSize": total_size,
+        "blobCount": len(blobs),
+        "volumeBudget": VOLUME_BUDGET,
+        "volumeSize": VOLUME_BUDGET,
+        "volumes": len(volumes),
+        "entries": volumes,
+    }
+
+
+def _zip_tmp_root() -> Path:
+    """ZIP 打包临时目录：优先 /tmp（容器内），回退 /data，最后系统临时目录。"""
+    for candidate in ("/tmp/backup_v6_zip", "/data/backup_v6_zip"):
+        try:
+            p = Path(candidate)
+            p.mkdir(parents=True, exist_ok=True)
+            (p / ".wtest").touch(exist_ok=True)
+            (p / ".wtest").unlink(missing_ok=True)
+            return p
+        except Exception:
+            continue
+    import tempfile
+    p = Path(tempfile.gettempdir()) / "backup_v6_zip"
+    p.mkdir(parents=True, exist_ok=True)
+    return p
+
+
+def build_volume_zip(set_id: str, volume: int) -> tuple[Path, int]:
+    """构建第 volume 卷 ZIP（含完整 manifest.json + 本卷 blob_<idx>.dat）。
+
+    扁平结构：小天才 zip.decompress 对子目录 ZIP 不可靠（v5 实测），
+    blob 文件统一命名 blob_<idx>.dat，idx 与 manifest.blobs 数组下标对应。
+    每卷都写一份 manifest.json，客户端任一卷即可拿到完整清单。
+    """
+    info = archive_info(set_id)
+    if volume < 0 or volume >= info["volumes"]:
+        raise ValueError("volume_out_of_range")
+    manifest = get_set_manifest(set_id)
+    blobs = manifest.get("blobs", []) or []
+
+    # 带 idx 的 manifest 副本
+    blobs_out = []
+    for i, entry in enumerate(blobs):
+        e = dict(entry) if isinstance(entry, dict) else {}
+        e["idx"] = i
+        blobs_out.append(e)
+    manifest_out = dict(manifest)
+    manifest_out["blobs"] = blobs_out
+    manifest_out["v6Volumes"] = info["volumes"]
+    manifest_out["v6CurrentVolume"] = volume
+
+    tmp_root = _zip_tmp_root()
+    # 唯一文件名：并发请求（双设备恢复同一备份点 / 重试重叠）互不覆盖，
+    # 各自的流式传输结束时删各自的临时文件
+    import uuid as _uuid
+    tmp_path = tmp_root / f"{set_id}_v{volume}_{_uuid.uuid4().hex[:8]}.zip"
+
+    entry_spec = info["entries"][volume]
+    with zipfile.ZipFile(tmp_path, "w", zipfile.ZIP_DEFLATED) as zf:
+        zf.writestr("manifest.json", json.dumps(manifest_out, ensure_ascii=False))
+        for i in range(entry_spec["startIdx"], entry_spec["endIdx"]):
+            b = blobs[i]
+            if not isinstance(b, dict):
+                continue
+            key = str(b.get("key") or "")
+            if not key:
+                continue
+            blob_path = get_blob_local_path(key)
+            if not blob_path or not blob_path.exists():
+                zf.writestr(f"blob_{i}.dat.MISSING", f"blob {key} missing")
+                continue
+            zf.write(blob_path, f"blob_{i}.dat")
+
+    return tmp_path, tmp_path.stat().st_size
+
+
+def zip_iter_chunks(zip_path: Path, chunk_size: int = 64 * 1024):
+    """流式读取 ZIP 文件（传完后删除临时文件）。"""
+    async def _gen():
+        try:
+            with open(zip_path, "rb") as f:
+                while True:
+                    chunk = f.read(chunk_size)
+                    if not chunk:
+                        break
+                    yield chunk
+        except Exception as e:
+            logger.error("[v6_zip] iter chunks failed: %s", e)
+            raise
+        finally:
+            try:
+                zip_path.unlink(missing_ok=True)
+            except Exception:
+                pass
+    return _gen()
+
+
+# ===== GC（垃圾回收）=====
+
+def gc_orphan_blobs() -> dict:
+    """清理 refs=0 且上传超过宽限期的 blob（本地 + Hub + DB）。"""
+    cutoff = _now() - GC_BLOB_GRACE_SEC
+    with get_conn() as conn:
+        rows = conn.execute(
+            "SELECT key, size FROM backup_v6_blobs WHERE refs <= 0 AND uploaded_at < ?",
+            (cutoff,),
+        ).fetchall()
+    freed = 0
+    freed_bytes = 0
+    hub_deletes: list[str] = []
+    for r in rows:
+        key = str(r["key"])
+        freed += 1
+        freed_bytes += int(r["size"])
+        try:
+            with get_conn() as conn:
+                conn.execute("DELETE FROM backup_v6_blobs WHERE key = ?", (key,))
+            p = _blob_local_path(key)
+            if p.exists():
+                p.unlink(missing_ok=True)
+            try:
+                if p.parent.exists() and not any(p.parent.iterdir()):
+                    p.parent.rmdir()
+            except OSError:
+                pass
+            if is_hub_enabled():
+                hub_deletes.append(key)
+        except Exception as e:
+            logger.warning("[v6_gc] delete blob failed key=%s: %s", key[:8], e)
+    if hub_deletes:
+        def _del_hub():
+            for key in hub_deletes:
+                try:
+                    delete_from_hub(NS_BLOBS, f"{key[:2]}/{key}")
+                except Exception:
+                    pass
+        try:
+            loop = asyncio.get_running_loop()
+            loop.run_in_executor(None, _del_hub)
+        except RuntimeError:
+            _del_hub()
+    return {"freedBlobs": freed, "freedBytes": freed_bytes}
+
+
+_gc_task: "Optional[asyncio.Task]" = None
+
+
+def start_gc_task() -> None:
+    global _gc_task
+    if _gc_task is not None:
+        return
+
+    async def _loop():
+        while True:
+            await asyncio.sleep(GC_INTERVAL_SEC)
+            try:
+                result = await asyncio.to_thread(gc_orphan_blobs)
+                if result["freedBlobs"] > 0:
+                    logger.info("[v6_gc] freed %d blobs (%d bytes)",
+                                result["freedBlobs"], result["freedBytes"])
+            except Exception as e:
+                logger.warning("[v6_gc] failed: %s", e)
+
+    try:
+        _gc_task = asyncio.create_task(_loop())
+    except RuntimeError:
+        pass
+
+
+async def stop_gc_task() -> None:
+    global _gc_task
+    if _gc_task is not None:
+        _gc_task.cancel()
+        try:
+            await _gc_task
+        except asyncio.CancelledError:
+            pass
+        _gc_task = None
+
+
+# ===== Hub 同步（HF Space 重建后恢复备份点元数据）=====
+
+def sync_sets_from_hub() -> int:
+    """从 Hub 的 backup_v6_sets 命名空间拉回所有 manifest，重建 DB 记录。
+
+    HF Space 重建后本地 DB 丢失，但 Hub 上 manifest 仍在。
+    blob 内容本体按需从 Hub 拉回（get_blob_local_path 内置回源）。
+    返回同步的 set 数。
+    """
+    if not is_hub_enabled():
+        return 0
+    synced = 0
+    try:
+        keys = list_hub_keys(NS_SETS)
+    except Exception as e:
+        logger.warning("[v6_sync] list hub keys failed: %s", e)
+        return 0
+    for sub_key in keys:
+        # sub_key 形如 <user_id>/<set_id>.json
+        parts = str(sub_key).split("/")
+        if len(parts) != 2 or not parts[0] or not parts[1].endswith(".json"):
+            continue
+        user_id, filename = parts[0], parts[1]
+        set_id = filename[:-5]
+        if not set_id:
+            continue
+        try:
+            with get_conn() as conn:
+                exists = conn.execute(
+                    "SELECT 1 FROM backup_v6_sets WHERE set_id = ?", (set_id,)
+                ).fetchone()
+            if exists:
+                continue
+            data = download_bytes(NS_SETS, f"{user_id}/{set_id}.json")
+            if not data:
+                continue
+            manifest = json.loads(data.decode("utf-8"))
+            blobs = manifest.get("blobs", []) or []
+            keys_in_set = [str(b.get("key") or "") for b in blobs if isinstance(b, dict)]
+            total_size = sum(int(b.get("size") or 0) for b in blobs)
+            checksum = str(manifest.get("checksum") or "")
+            created_at = int(manifest.get("exportedAt") or _now())
+            # blob DB 记录补插（本地/Hub 有内容即可，refs 由各 set 累加）
+            with get_conn() as conn:
+                for i, k in enumerate(dict.fromkeys(keys_in_set)):
+                    if not k:
+                        continue
+                    size = 0
+                    mime = ""
+                    for b in blobs:
+                        if isinstance(b, dict) and str(b.get("key") or "") == k:
+                            size = int(b.get("size") or 0)
+                            mime = _guess_mime(str(b.get("path") or ""))
+                            break
+                    conn.execute(
+                        "INSERT OR IGNORE INTO backup_v6_blobs(key, size, mime, refs, uploaded_at) "
+                        "VALUES(?, ?, ?, 0, ?)",
+                        (k, size, mime, created_at),
+                    )
+                conn.execute(
+                    "INSERT OR IGNORE INTO backup_v6_sets(set_id, user_id, alias, manifest, "
+                    "manifest_sha256, total_size, blob_count, created_at, is_latest, prev_set_id, mode, device) "
+                    "VALUES(?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?)",
+                    (
+                        set_id, user_id, str(manifest.get("alias") or "")[:200],
+                        json.dumps(manifest, ensure_ascii=False), checksum,
+                        total_size, len(blobs), created_at,
+                        str(manifest.get("prevSetId") or ""),
+                        str(manifest.get("mode") or "")[:32],
+                        str(manifest.get("device") or "")[:128],
+                    ),
+                )
+                for k in dict.fromkeys(keys_in_set):
+                    if k:
+                        conn.execute(
+                            "UPDATE backup_v6_blobs SET refs = refs + 1 WHERE key = ?", (k,)
+                        )
+            # 重建 is_latest（最新 created_at 的 set）
+            with get_conn() as conn:
+                newest = conn.execute(
+                    "SELECT set_id FROM backup_v6_sets WHERE user_id = ? ORDER BY created_at DESC LIMIT 1",
+                    (user_id,),
+                ).fetchone()
+                conn.execute(
+                    "UPDATE backup_v6_sets SET is_latest = 0 WHERE user_id = ?", (user_id,)
+                )
+                if newest:
+                    conn.execute(
+                        "UPDATE backup_v6_sets SET is_latest = 1 WHERE set_id = ?",
+                        (str(newest["set_id"]),),
+                    )
+            _write_manifest_to_local(user_id, set_id, manifest)
+            synced += 1
+        except Exception as e:
+            logger.warning("[v6_sync] sync set %s failed: %s", set_id[:8], e)
+    if synced:
+        logger.info("[v6_sync] restored %d sets from Hub", synced)
+    return synced
+
+
+async def _sync_sets_background() -> None:
+    try:
+        await asyncio.to_thread(sync_sets_from_hub)
+    except Exception as e:
+        logger.warning("[v6_sync] background sync failed: %s", e)
+
+
+# ===== v5 遗留数据清理 =====
+
+def purge_legacy_v5(user_id: Optional[str] = None) -> dict:
+    """清理 v5 遗留数据（已被污染、设计缺陷无法正常使用）。
+
+    - user_id 指定：只清该用户的 v5 备份集 + 全部 refs=0 的 v5 blob + Hub 上该用户的 manifest
+    - user_id 为 None：清空全部 v5 数据（表 + 本地目录 + Hub 命名空间）
+
+    ★ 本函数会在工作线程（asyncio.to_thread）中执行，绝不调用任何含
+      asyncio.create_task / ensure_future 的函数（v5 的 delete_set 会在末尾
+      create_task 触发 GC，线程内无事件循环直接 RuntimeError），
+      全部用同步 SQL + 文件操作实现。
+
+    返回 {deletedSets, deletedBlobs, freedBytes}。
+    """
+    from ..hf_storage import _local_path
+
+    deleted_sets = 0
+    deleted_blobs = 0
+    freed_bytes = 0
+
+    def _v5_blob_path(key: str) -> Path:
+        return _local_path("backup_blobs_v5", f"{key[:2]}/{key}")
+
+    if user_id:
+        # 用户级：直接 SQL 批量删（不走 v5 delete_set，避免其内部 create_task）
+        with get_conn() as conn:
+            rows = conn.execute(
+                "SELECT set_id, manifest FROM backup_sets WHERE user_id = ?", (user_id,)
+            ).fetchall()
+        for r in rows:
+            set_id = str(r["set_id"])
+            # 解析 manifest 拿 blob keys，直接减 refs（含 DB/本地/Hub 三处清理）
+            keys: list[str] = []
+            try:
+                m = json.loads(r["manifest"])
+                keys = [str(b.get("key") or "") for b in m.get("blobs", []) if isinstance(b, dict)]
+            except Exception:
+                keys = []
+            try:
+                with get_conn() as conn:
+                    for k in dict.fromkeys(keys):
+                        conn.execute(
+                            "UPDATE backup_blobs SET refs = MAX(refs - 1, 0) WHERE key = ?", (k,)
+                        )
+                    conn.execute("DELETE FROM backup_sets WHERE set_id = ?", (set_id,))
+                deleted_sets += 1
+            except Exception as e:
+                logger.warning("[v6_purge] delete v5 set failed set=%s: %s", set_id[:8], e)
+            # 删本地 + Hub 的 manifest
+            try:
+                mp = _local_path("backup_sets_v5", f"{user_id}/{set_id}.json")
+                if mp.exists():
+                    mp.unlink(missing_ok=True)
+                if is_hub_enabled():
+                    delete_from_hub("backup_sets_v5", f"{user_id}/{set_id}.json")
+            except Exception:
+                pass
+        # 清理全部 refs=0 的 v5 blob（无论宽限期——v5 整体废弃，同步删）
+        with get_conn() as conn:
+            orphans = conn.execute(
+                "SELECT key, size FROM backup_blobs WHERE refs <= 0"
+            ).fetchall()
+        for r in orphans:
+            key = str(r["key"])
+            try:
+                with get_conn() as conn:
+                    conn.execute("DELETE FROM backup_blobs WHERE key = ?", (key,))
+                    conn.execute("DELETE FROM backup_blob_uploaders WHERE key = ?", (key,))
+                p = _v5_blob_path(key)
+                if p.exists():
+                    p.unlink(missing_ok=True)
+                if is_hub_enabled():
+                    try:
+                        delete_from_hub("backup_blobs_v5", f"{key[:2]}/{key}")
+                    except Exception:
+                        pass
+                deleted_blobs += 1
+                freed_bytes += int(r["size"])
+            except Exception as e:
+                logger.warning("[v6_purge] delete v5 blob failed key=%s: %s", key[:8], e)
+        # 删 Hub 上该用户的其余 v5 manifest（防御本地 DB 已缺记录的情况）
+        if is_hub_enabled():
+            try:
+                for sub in list_hub_keys("backup_sets_v5"):
+                    if str(sub).split("/")[0] == user_id:
+                        try:
+                            delete_from_hub("backup_sets_v5", sub)
+                        except Exception:
+                            pass
+            except Exception:
+                pass
+    else:
+        # 全局清理：清表 + 清本地目录 + 清 Hub 命名空间
+        with get_conn() as conn:
+            rows = conn.execute("SELECT COUNT(*) AS c FROM backup_sets").fetchone()
+            deleted_sets = int(rows["c"]) if rows else 0
+            conn.execute("DELETE FROM backup_sets")
+            conn.execute("DELETE FROM backup_blobs")
+            conn.execute("DELETE FROM backup_blob_uploaders")
+        # 本地文件
+        import shutil
+        for ns in ("backup_blobs_v5", "backup_sets_v5"):
+            try:
+                root = _local_path(ns, "_").parent
+                if root.exists():
+                    shutil.rmtree(root, ignore_errors=True)
+            except Exception:
+                pass
+        # Hub 命名空间
+        if is_hub_enabled():
+            for ns in ("backup_blobs_v5", "backup_sets_v5"):
+                try:
+                    for sub in list_hub_keys(ns):
+                        try:
+                            delete_from_hub(ns, sub)
+                        except Exception:
+                            pass
+                except Exception:
+                    pass
+        deleted_blobs = -1  # 全局模式下不精确统计
+
+    return {"deletedSets": deleted_sets, "deletedBlobs": deleted_blobs, "freedBytes": freed_bytes}