feat(shard): 接入 Redis 协调存储与观测链路

Author: TogetsuDo

bot_worker.py

@@ -8,6 +8,7 @@
     PORT=8090
 """
 
+import asyncio
 import os
 
 os.environ.setdefault("PALLAS_SHARD_ENABLED", "true")
@@ -64,6 +65,15 @@ def pin_worker_listen_port() -> None:
 register_onebot_v11_custom_events()
 
 
+async def _ensure_worker_voices_background() -> None:
+    try:
+        ok = await ensure_voices()
+        if not ok:
+            nonebot.logger.warning("bot_worker: voice ensure failed or incomplete")
+    except Exception as err:
+        nonebot.logger.warning("bot_worker: voice ensure failed: {}", err)
+
+
 @driver.on_startup
 async def startup():
     await init_db()
@@ -97,7 +107,7 @@ async def startup():
     if coord_redis_enabled():
         nonebot.logger.info("bot_worker: cross-process claims via Redis ({})", resolve_coord_redis_url())
     start_shard_coord_worker_watcher()
-    await ensure_voices()
+    asyncio.create_task(_ensure_worker_voices_background(), name="worker_ensure_voices")
 
 
 @driver.on_shutdown

docs/architecture/bot_process_sharding.md

@@ -179,25 +179,25 @@ WebUI：`GET /pallas/api/shard-registry` 查看分片与 QQ 归属（需控制
 
 ## 跨 worker 协调（共享 `data/pallas_shard/`）
 
-| 能力 | 目录 / 模块 | 说明 |
+| 能力 | Redis 键 / 模块 | 说明 |
 |------|-------------|------|
-| 牛牛报数顺序 | `coord/bot_count/` | 各片登记 → 收集窗口 → 最小 QQ finalize 顺序 |
-| 决斗 QTE 代答 | `coord/duel_qte/` | 主持片写会话，应答牛所在片 watcher 回写 |
-| 同群决斗互斥 | `coord/duel_group/` | 跨片占用 |
-| 指定牛代发 API | `coord/bot_action/` | 跨片请求 + watcher |
-| 复读近期语料 | `coord/repeater_buffer/` + Redis Pub/Sub | ingress 获胜片广播；有 Redis 时订阅即时合并，否则文件轮询回退 |
-| 群级短占位 | `coord/group_gate/` | 含 `draw` owned gate |
+| 牛牛报数顺序 | `pallas:coord:bot_count:*` | 各片登记 → 收集窗口 → 最小 QQ finalize 顺序 |
+| 决斗 QTE 代答 | `pallas:duel_qte:session:*` + Pub/Sub | 主持片写会话，应答牛所在片 listener 回写 |
+| 同群决斗互斥 | `pallas:coord:duel_group:*` | 跨片占用 |
+| 指定牛代发 API | `pallas:coord:bot_action:*` + Pub/Sub | 跨片请求 + listener |
+| 复读近期语料 | `pallas:repeater_buffer` + Pub/Sub | ingress 获胜片广播并即时合并 |
+| 群级短占位 | `pallas:coord:group_gate:*` | 含 `draw` owned gate |
 | 在线态 / WebUI | `worker_presence.json` | hub 读各 worker 连接 |
 | 控制台指标 | `stats/worker-{N}.json` | hub 合并展示 |
 | registry / fleet | `registry.json` + mtime | `shard_data_sync` 失效缓存 |
 
-worker 由 `src/platform/shard/coord/worker_poll.py` 轮询 `duel_qte`、`bot_action` 待办。
+worker 由 `src/platform/shard/coord/worker_poll.py` 启动 Redis listener，监听 `duel_qte`、`bot_action` 与 repeater 缓冲广播。
 
 ## WebUI 与日志（hub）
 
 - `GET /pallas/api/bots`：读 **worker_presence.json**（hub 无反向 WS）。
 - `GET /pallas/api/logs`：合并 `data/pallas_shard/logs/hub.log` 与各 `worker-*.log` 尾行（`sharded_logs: true`）。
-- `GET /pallas/api/shard-observability`：合并各 worker `stats/worker-*.json` 的 **ingress 命中率**、**coord 积压**（实时扫描 `coord/`）、**PG 连接池粗算**；**WebUI 首页**在分片模式下展示「分片可观测」面板。
+- `GET /pallas/api/shard-observability`：合并各 worker `stats/worker-*.json` 的 **ingress 命中率**、**coord 积压**（实时统计 Redis 键空间）、**PG 连接池粗算**；**WebUI 首页**在分片模式下展示「分片可观测」面板。
 - `./scripts/run_sharded_bot.sh status`：输出 **分片可观测** 摘要（同 API 数据源）。
 - `uv run python scripts/shard_observability_snapshot.py`：完整 JSON；`scripts/shard_observability_status.py`：终端摘要。
 - `plugin-run-stats` / `message-stats`：合并各 worker 的 `stats/worker-*.json`；ERROR 日志合并 worker/hub 落盘文件。

docs/plugins/maa/README.md

@@ -45,7 +45,7 @@
 | 现象 | 处理 |
 | --- | --- |
 | 未检测到轮询 | MAA 端点不可达或 URL 错误；分片须 hub `maa_public_base_url` 且各 worker 共用 `data/` |
-| 状态有待拉取、MAA 无任务 | 分片时队列在 `data/pallas_shard/coord/maa_pending/`；须 hub 能访问各 worker 端口 |
+| 状态有待拉取、MAA 无任务 | 分片时队列走 Redis `pallas:coord:maa_pending:*`；须 hub 能访问各 worker 端口并保证 Redis 可用 |
 | 下发后无任务 | 未绑定或用户标识符非 QQ；查 `牛牛MAA状态` |
 | 队列有、MAA 无 | 设备 id 与「当前选用」不一致；可清空队列重试 |
 | 截图失败 | 调大反代 `client_max_body_size` |

scripts/detect_shard_redis.py

@@ -48,7 +48,7 @@ def main() -> int:
     pkg = "yes" if redis_package_installed() else "no"
     reachable = "yes" if url and coord_redis_enabled() else "no"
     active = "yes" if reachable == "yes" else "no"
-    backend = "redis" if active == "yes" else "file"
+    backend = "redis" if active == "yes" else "unavailable"
 
     if args.status:
         print(f"policy={mode}")

scripts/prune_shard_coord.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""一次性清理 coord 陈旧 JSON（bot_action / duel_qte 等）。分片运行时可在 hub 或运维机执行。"""
+"""coord 清理脚本（coord 已 Redis 化，本脚本仅输出快照供运维兼容）。"""
 
 from __future__ import annotations
 
@@ -14,51 +14,29 @@
     sys.path.insert(0, str(REPO_ROOT))
 
 
-def prune_all_done_bot_action() -> int:
-    from src.platform.shard.coord.bot_action import _coord_dir, _read
-
-    removed = 0
-    for path in _coord_dir().glob("*.json"):
-        if ".lock" in path.name:
-            continue
-        row = _read(path)
-        if isinstance(row, dict) and row.get("done"):
-            try:
-                path.unlink(missing_ok=True)
-                removed += 1
-            except OSError:
-                pass
-    return removed
-
-
 async def main() -> int:
-    parser = argparse.ArgumentParser(description="清理 data/pallas_shard/coord 陈旧 JSON")
+    parser = argparse.ArgumentParser(description="coord Redis 模式快照（legacy 文件清理已停用）")
     parser.add_argument(
         "--purge-done",
         action="store_true",
-        help="删除全部 bot_action done 文件（运维一次性减压，不影响进行中的 open）",
+        help="兼容旧参数；Redis 模式下无文件可清理",
+    )
+    parser.add_argument(
+        "--live-scan",
+        action="store_true",
+        help="显式扫描 Redis 键空间；默认仅输出轻量快照",
     )
     args = parser.parse_args()
-    from src.platform.shard.coord.bot_action import prune_stale_bot_action_files
-    from src.platform.shard.coord.bot_count import prune_stale_bot_count_files
-    from src.platform.shard.coord.cage_duel import prune_stale_cage_duel_files
-    from src.platform.shard.coord.duel_group import prune_stale_duel_group_files
-    from src.platform.shard.coord.duel_qte import prune_stale_duel_qte_files
-    from src.platform.shard.coord.repeater_buffer import prune_stale_repeater_buffer_files
     from src.platform.shard.coord_pending import coord_pending_snapshot_sync
 
-    before = coord_pending_snapshot_sync()
-    purged_done = prune_all_done_bot_action() if args.purge_done else 0
-    bot_stats = await prune_stale_bot_action_files()
-    if purged_done:
-        bot_stats["purged_all_done"] = purged_done
-    await prune_stale_duel_qte_files()
-    await prune_stale_repeater_buffer_files()
-    await prune_stale_cage_duel_files()
-    await prune_stale_bot_count_files()
-    await prune_stale_duel_group_files()
-    after = coord_pending_snapshot_sync()
-    out = {"before": before, "bot_action": bot_stats, "after": after}
+    snap = coord_pending_snapshot_sync(live=bool(args.live_scan))
+    out = {
+        "storage": snap.get("storage"),
+        "snapshot": snap,
+        "purge_done_requested": bool(args.purge_done),
+        "live_scan_requested": bool(args.live_scan),
+        "note": "coord 数据存 Redis，无 JSON 文件轮询/prune",
+    }
     json.dump(out, sys.stdout, ensure_ascii=False, indent=2)
     sys.stdout.write("\n")
     return 0

scripts/run_sharded_bot.sh

@@ -39,6 +39,7 @@ WAIT_PORTS_SCRIPT="${SCRIPT_DIR}/wait_shard_worker_ports.py"
 SHARD_TEST_SCRIPT="${SCRIPT_DIR}/shard_test_worker.py"
 DETECT_REDIS_SCRIPT="${SCRIPT_DIR}/detect_shard_redis.py"
 PORT_RELEASE_TIMEOUT="${PALLAS_SHARD_PORT_RELEASE_TIMEOUT:-60}"
+FORCE_STOP=0
 PID_WORKER_TEST="${RUN_DIR}/worker-test.pid"
 TEST_SHARD_ID="${PALLAS_SHARD_TEST_ID:-99}"
 
@@ -96,6 +97,7 @@ usage() {
   --workers-only       仅操作生产 worker（不停/不启 hub；start 时只拉起缺失进程，restart 时全量重启）
   --scale-only         扩容模式：不重分配端口、不同步协议端 ws_url（已有 worker 在跑时 start --workers-only 自动启用）
   --hub-only           仅操作 hub 控制台（不停/不启 worker；restart 时常用）
+  --force              stop/restart 时 SIGKILL 强杀并跳过端口长等待（加快重启）
   --dry-run            只显示将要执行的命令，不真正启动
   -h, --help           显示本帮助
 
@@ -330,6 +332,13 @@ stop_one() {
   fi
   local pid
   pid="$(cat "${pidfile}")"
+  if [[ "${FORCE_STOP}" -eq 1 ]]; then
+    kill -KILL "${pid}" 2>/dev/null || true
+    pkill -KILL -P "${pid}" 2>/dev/null || true
+    rm -f "${pidfile}"
+    echo "  · ${label}：已强制停止"
+    return 0
+  fi
   kill -TERM "${pid}" 2>/dev/null || true
   local i=0
   while kill -0 "${pid}" 2>/dev/null && [[ "${i}" -lt 30 ]]; do
@@ -338,6 +347,7 @@ stop_one() {
   done
   if kill -0 "${pid}" 2>/dev/null; then
     kill -KILL "${pid}" 2>/dev/null || true
+    pkill -KILL -P "${pid}" 2>/dev/null || true
   fi
   rm -f "${pidfile}"
   echo "  · ${label}：已停止"
@@ -347,8 +357,15 @@ stop_one() {
 stop_orphan_shard_processes() {
   local pat
   for pat in "${REPO_ROOT}/.venv/bin/python3 bot_hub.py" "${REPO_ROOT}/.venv/bin/python3 bot_worker.py"; do
-    pkill -TERM -f "${pat}" 2>/dev/null || true
+    if [[ "${FORCE_STOP}" -eq 1 ]]; then
+      pkill -KILL -f "${pat}" 2>/dev/null || true
+    else
+      pkill -TERM -f "${pat}" 2>/dev/null || true
+    fi
   done
+  if [[ "${FORCE_STOP}" -eq 1 ]]; then
+    return 0
+  fi
   sleep 1
   for pat in "${REPO_ROOT}/.venv/bin/python3 bot_hub.py" "${REPO_ROOT}/.venv/bin/python3 bot_worker.py"; do
     pkill -KILL -f "${pat}" 2>/dev/null || true
@@ -357,13 +374,21 @@ stop_orphan_shard_processes() {
 
 stop_orphan_worker_processes() {
   local pat="${REPO_ROOT}/.venv/bin/python3 bot_worker.py"
+  if [[ "${FORCE_STOP}" -eq 1 ]]; then
+    pkill -KILL -f "${pat}" 2>/dev/null || true
+    return 0
+  fi
   pkill -TERM -f "${pat}" 2>/dev/null || true
   sleep 1
   pkill -KILL -f "${pat}" 2>/dev/null || true
 }
 
 stop_orphan_hub_processes() {
   local pat="${REPO_ROOT}/.venv/bin/python3 bot_hub.py"
+  if [[ "${FORCE_STOP}" -eq 1 ]]; then
+    pkill -KILL -f "${pat}" 2>/dev/null || true
+    return 0
+  fi
   pkill -TERM -f "${pat}" 2>/dev/null || true
   sleep 1
   pkill -KILL -f "${pat}" 2>/dev/null || true
@@ -482,6 +507,11 @@ count_running_production_worker_ids() {
 
 wait_worker_ports_released() {
   local workers="$1"
+  if [[ "${FORCE_STOP}" -eq 1 ]]; then
+    echo "  · worker 端口：强制模式，短等待 3s"
+    sleep 3
+    return 0
+  fi
   if [[ "${DRY_RUN}" -eq 1 ]]; then
     echo "  · worker 端口：将等待释放后再启动（预览）"
     return 0
@@ -1024,7 +1054,7 @@ cmd_stop() {
     echo "  生产 worker 已处理完毕（测试 worker-test 未停止）。"
     return 0
   fi
-  print_title "Pallas-Bot 分片模式 · 停止"
+  print_title "Pallas-Bot 分片模式 · 停止$([[ "${FORCE_STOP}" -eq 1 ]] && echo '（强制）')"
   stop_one "worker-test" "测试 worker-test"
   stop_production_workers
   stop_hub_process
@@ -1345,6 +1375,10 @@ while [[ $# -gt 0 ]]; do
       HUB_ONLY=1
       shift
       ;;
+    --force)
+      FORCE_STOP=1
+      shift
+      ;;
     -h|--help)
       usage
       exit 0

scripts/shard_observability_status.py

@@ -50,15 +50,17 @@ def main() -> int:
         f"早丢弃={int(ing.get('early_fleet', 0)) + int(ing.get('early_not_at_target', 0))}"
     )
     print(
-        f"coord JSON                total={coord.get('total_json', 0)}  "
+        f"coord Redis              keys={coord.get('total_json', 0)}  "
         f"actionable={coord.get('actionable_total', coord.get('bot_action_open', 0))}  "
         f"historical={coord.get('historical_retained', 0)}  "
         f"bot_action_open={coord.get('bot_action_open', 0)}  "
         f"stale_open={coord.get('bot_action_stale_open', 0)}"
     )
+    if coord.get("scan_skipped"):
+        print("coord live scan          skipped（默认不扫 Redis，全量排查用 prune_shard_coord.py --live-scan）")
     ba = (coord.get("by_dir") or {}).get("bot_action", 0)
     if ba:
-        print(f"coord bot_action 文件     {ba}  （可 prune_shard_coord.py --purge-done）")
+        print(f"coord bot_action 键       {ba}  （prune_shard_coord.py 仅输出快照）")
 
     peak = pg.get("estimated_pg_connections_peak")
     proc = pg.get("estimated_processes")