From 8b482e7a30990875f9e20b9d64b7bf338db1ea73 Mon Sep 17 00:00:00 2001 From: caoqianming Date: Mon, 6 Jul 2026 14:51:39 +0800 Subject: [PATCH] =?UTF-8?q?feat(broker):=20=E5=A4=96=E7=BD=AE=20Redis=20pu?= =?UTF-8?q?b/sub=E2=80=94=E2=80=94=E8=93=9D=E7=BB=BF=E8=B7=A8=E5=AE=9E?= =?UTF-8?q?=E4=BE=8B=20SSE/cancel=20=E5=8F=AF=E8=BE=BE(bump=200.42.0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用户量上升重评落地(DESIGN §7.0):web/broker.py 拆 LocalRunBroker(原进程内 实现,默认)/ RedisRunBroker 同接口双实现,ZCBOT_REDIS_URL env 开关。 - event:PUBLISH zcbot:ev: + 单条 async pubsub reader 路由本地订阅 queue (Redis 只管跨进程一跳,fan-out 最后一公里仍在进程内) - done:SETEX 60s key + publish 双通道;订阅先挂 channel 再查 key,晚到不漏 - cancel:SETEX/GET/DEL key,loop chunk 间 poll,跨实例可达——蓝绿切换窗口 「刷新丢直播 / 停止失灵」两个边缘消失;稳态双实例分流只差 nginx 配置 - 容错:启动 ping 不通 fail-fast(同 sandbox init 纪律);运行中失败 10s 节流 log + 降级丢帧,run 不受影响;reader 1s 退避重连并重挂活跃 channel - lifespan finally 加 broker.shutdown();sinks.py 解开对旧类名的 import - tests/test_broker_redis.py 9 例(fakeredis:跨实例 emit/cancel、晚订阅 done、 降级不抛、fail-fast);requirements 加 redis/fakeredis - RUN.md:env 段 + 蓝绿节 + bootstrap(apt install redis-server)+ 故障兜底 3 行 Co-Authored-By: Claude Fable 5 --- DESIGN.md | 4 +- PROGRESS.md | 3 +- RUN.md | 18 ++- core/__init__.py | 2 +- requirements.txt | 5 + tests/test_broker_redis.py | 177 ++++++++++++++++++++++ web/app.py | 7 + web/broker.py | 295 ++++++++++++++++++++++++++++++++++--- web/sinks.py | 6 +- 9 files changed, 487 insertions(+), 30 deletions(-) create mode 100644 tests/test_broker_redis.py diff --git a/DESIGN.md b/DESIGN.md index 50708b9..28412c0 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -127,7 +127,7 @@ Session = 消息列表,ORM 直写 PG `messages`(append-only,jsonb 存 LiteLLM **无感部署(蓝绿,0.41)**:生产双 systemd 实例(`ZCBOT_INSTANCE=blue/green`)+ nginx upstream 切流(对外仍 8765),取代单实例 restart 的 503 窗口。双实例并存的互踩由三件事消解:`tasks.run_owner`(0020)让 reaper 只收自己色、sandbox 容器名/label 带实例色各管各的、微信长轮询 PG advisory lock 选主;定时任务 claim 本就 SKIP LOCKED 天然安全。部署编排在 `deploy/update_bluegreen.sh`(流程/兜底见 RUN)。 -**broker 外置(Redis pub/sub)**:0.41 落地时 event/cancel broker 留在进程内,代价是切换窗口内"刷新看不到旧实例 run 直播 / 停止送达不到"两个边缘。**2026-07-06 重评(用户量已非个位数):由"搁置到真实投诉"转为"推荐实施"** —— ①部署时总有 in-flight run,窗口边缘从偶发变常态;②单进程逼近天花板后,最近的扩容手段是稳态双实例同时接流量,broker 外置是它的硬前提(否则 POST 与 SSE 落不同实例直播全瞎)。选 Redis 不选 PG LISTEN/NOTIFY(token 级 delta 全过主库太吵:NOTIFY 全局队列 + 8KB payload 限制,把实时路径耦到 PG)、不选 nginx sticky hash(upstream 变更时 hash 重排,恰在部署窗口失效,治标)。方案:`ZCBOT_REDIS_URL` env 开关,不设即回退进程内(dev 零影响);done 语义改查 DB run_status + done key(天然跨实例);cancel 改 Redis key poll。Redis 不解决的:run 本体仍绑进程(进程死 run 死,drain/reaper 不变)、线程池上限(调 `ZCBOT_RUN_MAX_WORKERS` 的事)。 +**broker 外置(Redis pub/sub,✅ 0.42)**:0.41 落地时 event/cancel broker 留在进程内,代价是切换窗口内"刷新看不到旧实例 run 直播 / 停止送达不到"两个边缘。2026-07-06 重评(用户量已非个位数)决定实施 —— ①部署时总有 in-flight run,窗口边缘从偶发变常态;②单进程逼近天花板后,最近的扩容手段是稳态双实例同时接流量,broker 外置是它的硬前提(否则 POST 与 SSE 落不同实例直播全瞎)。选 Redis 不选 PG LISTEN/NOTIFY(token 级 delta 全过主库太吵:NOTIFY 全局队列 + 8KB payload 限制,把实时路径耦到 PG)、不选 nginx sticky hash(upstream 变更时 hash 重排,恰在部署窗口失效,治标)。实现(`web/broker.py`,LocalRunBroker/RedisRunBroker 同接口鸭子类型):`ZCBOT_REDIS_URL` env 开关,不设即进程内(dev 零影响);event 走 `PUBLISH zcbot:ev:` + 单条 async pubsub reader 路由到本地订阅 queue(Redis 只管跨进程一跳,fan-out 最后一公里仍在进程内);done = SETEX key(60s)+ publish 双通道,订阅先挂 channel 再查 key 不漏;cancel = SETEX/GET/DEL key,loop 在 chunk 间 poll(localhost ~0.1ms)。容错纪律:启动 ping 不通 fail-fast(同 sandbox init),运行中失败 10s 节流 log + 降级丢帧,reader 1s 退避重连。Redis 不解决的:run 本体仍绑进程(进程死 run 死,drain/reaper 不变)、线程池上限(调 `ZCBOT_RUN_MAX_WORKERS` 的事)。 ### 7.1 心智模型:Task 一等公民 + Dir 文件副视图 @@ -264,7 +264,7 @@ scheduled_jobs(§8.5) channel_bindings(§8.7,判别列+JSONB) ### 8.4 运维监控 / 无感更新(监控 ✅ / 换版 design) -**优雅 drain 已是单实例上限**(SIGTERM 拒新 run + 等收尾);先撞的瓶颈是线程池(每活跃 run 占 1 线程)。落地排序:① 轻量监控(显式 executor + 60s `[stats]` 周期日志——要历史峰值不是快照);② 按数据决策扩容;③ `--reload` 缩 503 窗。**不做监控界面**:运维健康是少数标量,日志够;业务分析走 DB SQL。界面阶梯:日志 → /v1/stats → Grafana → 只读 dashboard,现停第一级。无感换版已由蓝绿落地(0.41,见 §7.0);broker 外置 Redis 2026-07-06 重评为推荐实施(§7.0),扩容路径:调大线程池 → 稳态双实例分流(broker 外置后纯 nginx 配置动作)。 +**优雅 drain 已是单实例上限**(SIGTERM 拒新 run + 等收尾);先撞的瓶颈是线程池(每活跃 run 占 1 线程)。落地排序:① 轻量监控(显式 executor + 60s `[stats]` 周期日志——要历史峰值不是快照);② 按数据决策扩容;③ `--reload` 缩 503 窗。**不做监控界面**:运维健康是少数标量,日志够;业务分析走 DB SQL。界面阶梯:日志 → /v1/stats → Grafana → 只读 dashboard,现停第一级。无感换版已由蓝绿落地(0.41)、broker 外置 Redis 已实施(0.42,均见 §7.0);扩容路径:调大线程池 → 稳态双实例分流(broker 已外置,纯 nginx 配置动作)。 ### 8.5 定时任务(✅ 2026-06-18) diff --git a/PROGRESS.md b/PROGRESS.md index 6dff232..f733e63 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -2,7 +2,7 @@ > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。 -最后更新:2026-07-06(规模设定重评:broker 外置 Redis 转推荐实施,bump 0.41.2) +最后更新:2026-07-06(broker 外置 Redis 落地,bump 0.42.0) --- @@ -23,6 +23,7 @@ ### 2026-07 +- **07-06 / 0.42.0 broker 外置 Redis(用户量上升重评落地)**:`web/broker.py` 拆 LocalRunBroker(原进程内实现)/ RedisRunBroker 同接口双实现,`ZCBOT_REDIS_URL` env 开关(不设 = 进程内,dev/单实例零影响)。Redis 版:event 走 PUBLISH + 单条 async pubsub reader 路由本地 queue,done = SETEX(60s)+publish 双通道(订阅先挂 channel 再查 key 不漏),cancel = key GET poll 跨实例可达——蓝绿切换窗口「刷新丢直播/停止失灵」两边缘消失,稳态双实例分流从此只差 nginx 配置。容错:启动 ping fail-fast、运行中降级丢帧(10s 节流 log)、reader 1s 退避重连。`tests/test_broker_redis.py` 9 例(fakeredis,含跨实例/晚订阅/降级);deploy bootstrap 加 redis-server 步骤。关键判断:排除 PG LISTEN/NOTIFY(token delta 耦主库)与 nginx sticky(部署窗 hash 重排),Redis 只管跨进程一跳、fan-out 最后一公里仍在进程内。 - **07-06 / 0.41.1 蓝绿端口方案反转**:nginx 接管原对外端口 8765(客户端 / platform 嵌入零改动),实例挪 8766(blue)/ 8767(green) 只绑本机——对外契约(URL:8765)不动优先于"实例沿用旧端口"的省事;顺带把「切换窗口内对旧实例 run 点停止送达不到」补进已知边缘文档。 - **07-06 / 0.41.0 蓝绿无感部署(RUN.md B 档落地)**:生产部署零 503 窗口 + 在跑对话 SSE 直播不断。`zcbot@blue/green` 双 systemd 实例(模板 unit + `deploy/instances/*.env`)+ nginx upstream 单行文件切流,编排在 `deploy/update_bluegreen.sh`(复用 `update.sh --prepare-only`;切流后轮询 DB 等旧色 run 自然清空再 stop——直接 stop 会被 uvicorn `timeout_graceful_shutdown=5` 掐 SSE)。双实例并存互踩三修:`tasks.run_owner`(0020)reaper 只收自己色、sandbox 容器名/label 带实例色各清各的、微信 ClawBot 长轮询 PG advisory lock 选主(detach 连接持锁,断连即让位);定时任务 claim 本就 SKIP LOCKED 免改。关键判断:broker 刻意不出进程(不上 Redis),接受"切换瞬间刷新的用户看不到旧实例直播"这一边缘。单实例部署 `ZCBOT_INSTANCE` 未设全部退回原行为,`update.sh` 保留不废。 - **07-06 / 0.40.0 ppt 伪导出根治(三层)**:0.35.1 复发(模型照禁令里的"配方"装 cairosvg 手搓整页贴图导出)复盘——①平台层机检门落地(`core/pptx_guard.py` + loop 钩子:shell/run_python 新产出 .pptx 命中"≥80% 页整版位图 + 全 deck 零原生文本"即注入 `[产物机检 ERROR]` 当场逼返工;判定刻意保守,`tests/test_pptx_guard.py` 8 例);②禁令去菜谱:ppt/brief SKILL.md 否定式指令改"正面窄门"(唯一入口 + 报错就修/上报),删掉可照做的违规配方;③官方脚本 5 处 `pip install` hint 改环境说明(pptx_media/svg_preview/crop_images/fix_image_aspect/pptx_cli)——打印给 agent 的安装指令就是执行指令。详 DESIGN §8.9。 diff --git a/RUN.md b/RUN.md index 0fe1e51..8fc70e6 100644 --- a/RUN.md +++ b/RUN.md @@ -62,6 +62,12 @@ # SMTP_PASSWORD=<授权码/应用专用密码,非登录密码> # SMTP_FROM=you@qq.com # 可选,默认取 SMTP_USER # SMTP_FROM_NAME=总院科研辅助智能体 # 可选,发件人显示名,默"总院科研辅助智能体"(不暴露内部代号) + # broker 外置(DESIGN §7.0,蓝绿双实例跨进程 SSE/cancel):可选。设了 → event/cancel + # 走 Redis pub/sub(蓝绿两实例读同一份 .env,天然同 URL);不设 → 进程内 broker + # (dev / 单实例,零依赖不用起 redis)。设了但连不上 → 启动 fail-fast(报错见故障兜底)。 + # 蓝绿部署强烈建议设:消掉切换窗口「刷新丢直播 / 停止失灵」两个边缘,也是将来稳态 + # 双实例分流的前提。 + # ZCBOT_REDIS_URL=redis://127.0.0.1:6379/0 # 定时任务守护循环(DESIGN §8.5,随 web 进程起,plain-asyncio 仿 _disk_scanner): # ZCBOT_DISABLE_SCHEDULER=1 # 可选,整体关掉调度(对照 Claude Code CLAUDE_CODE_DISABLE_CRON) # ZCBOT_SCHEDULER_TICK_SECONDS=10 # 可选,扫描间隔,默 10s(只决定最坏延迟≤1tick,不影响会否漏) @@ -321,7 +327,9 @@ sudo bash /opt/zcbot/deploy/update_bluegreen.sh 脚本流程:`update.sh --prepare-only`(拉码 / pip / migration / build,全复用单实例脚本)→ 起空闲色实例 → healthz 验活(核对 `instance` 字段)→ 改写 upstream + nginx reload(**新请求全走新实例,零 503**)→ **轮询 DB 等旧色在跑 run 自然清空**(上限 `DRAIN_WAIT_SECONDS`,期间旧 SSE 继续直播不断流 —— 直接 stop 的话 uvicorn `timeout_graceful_shutdown=5` 会 5s 掐长连接)→ stop 旧实例(残留 run 走 SIGTERM drain / 协作式 cancel)→ 旧色 SIGKILL 残留的 running 行标 error → enable 新色 / disable 旧色(重启机器只拉起 live 色)。 -效果 vs 单实例 `update.sh`:**没有 503 窗口,在跑对话的 SSE 直播也不断**。已知边缘(接受,仅存在于切换后旧色 drain 的几分钟窗口内、且 run 恰好还在旧实例上):① 用户刷新页面 → SSE 连到新实例,broker 没有该 run 的事件,直播看不到(run 照常在旧实例跑完落库,完了刷新看历史);② 点「停止」→ cancel 信号进的是新实例的进程内 broker,旧实例收不到,等 run 自然收尾即可。两者根因同一个:event/cancel broker 是进程内的(`web/broker.py`)不跨实例。**2026-07-06 重评(用户量上升):broker 外置 Redis 已转为推荐实施**(消掉这两个边缘 + 为稳态双实例分流铺路,取舍与方案见 DESIGN §7.0);实施落地前此边缘仍在。 +效果 vs 单实例 `update.sh`:**没有 503 窗口,在跑对话的 SSE 直播也不断**。 + +**broker 外置(`ZCBOT_REDIS_URL`,0.42 起,蓝绿强烈建议设)**:不设时 event/cancel broker 是进程内的(`web/broker.py`),切换后旧色 drain 的几分钟窗口内有两个边缘 —— ① 刷新页面 SSE 连到新实例、看不到旧实例上 run 的直播(run 照常落库,完了刷新看历史);② 点「停止」送达不到旧实例上的 run(等自然收尾)。`.env` 设 `ZCBOT_REDIS_URL=redis://127.0.0.1:6379/0`(两实例共用同一 .env,自动同源)后 event/cancel 走 Redis pub/sub,**这两个边缘消失**,SSE 订阅与停止按钮跨实例可达;顺带打开将来稳态双实例分流(nginx upstream 挂两行)的大门。取舍与方案见 DESIGN §7.0;redis 运行中挂掉只降级直播丢帧(10s 节流 log),run 不受影响,恢复后 1s 内自愈。 蓝绿并存期的互踩问题已在代码层解决(单实例部署 `ZCBOT_INSTANCE` 未设,以下全部退回原行为): - **启动 reaper**:`tasks.run_owner`(migration 0020)记 run 归属实例,实例启动只收自己色(+ NULL 遗留)的孤儿 run,不误杀另一实例在跑的 @@ -346,6 +354,11 @@ sudo apt-get install -y nginx sudo cp deploy/nginx/zcbot.conf.example /etc/nginx/conf.d/zcbot.conf echo 'server 127.0.0.1:8766;' | sudo tee /etc/nginx/zcbot_upstream.conf +# 3.5) 装 redis(broker 外置,强烈建议)+ .env 加一行;默认绑 127.0.0.1 即可 +sudo apt-get install -y redis-server +redis-cli ping # 应回 PONG +echo 'ZCBOT_REDIS_URL=redis://127.0.0.1:6379/0' | sudo tee -a /opt/zcbot/.env + # 4) 停旧单实例服务(优雅 drain,释放 8765)→ 清无实例 label 的旧 sandbox 容器 # → 起 blue(8766)→ nginx 接管 8765。停旧到 nginx 起来之间是本次唯一的中断窗口 sudo systemctl disable --now zcbot @@ -841,6 +854,9 @@ sudo xfs_quota -x -c "limit -p bhard=10g zcbot_" /opt | 蓝绿切换后用户直播不动、但任务其实在跑 | 用户恰在切换窗口刷新页面,run 还在旧实例上而 SSE 连到了新实例(broker 进程内不跨实例)。run 照常跑完落库,等旧实例日志安静后让用户刷新看历史;非 bug,已知边缘 | | 蓝绿并存期微信 ClawBot 没反应 | 入站长轮询走 PG advisory lock 选主。两实例 journalctl 里找 `[wechat] inbound leader lock acquired`——都没有 → 检查 DB 连通;锁在已停实例上不可能(连接断锁即释),新实例最长 15s 接管 | | 蓝绿部署后旧色 sandbox 容器残留 | 正常:旧实例 stop 时 `shutdown_all` 只清自己色,失败/超时的靠下次该色启动兜底。手动清:`docker rm -f $(docker ps -aq -f label=zcbot.instance=<色>)` | +| 启动报 `ZCBOT_REDIS_URL 已设但 Redis 连不上` | redis-server 没起 / URL 错。`systemctl status redis-server` + `redis-cli ping` 验;或从 `.env` 删掉该行回退进程内 broker(蓝绿切换边缘会回来,见「broker 外置」段) | +| 日志刷 `[broker] redis publish failed`(10s 节流) | redis 运行中挂了。降级:直播丢帧、SSE 卡 ping,run 本身照常跑完落库;`systemctl restart redis-server` 恢复后 pubsub reader 1s 退避自动重连,无需重启 zcbot | +| 蓝绿切换窗口刷新丢直播 / 停止失灵(设了 ZCBOT_REDIS_URL 仍出现) | 两实例 journalctl 找 `[broker] redis pub/sub enabled` —— 没有 = 该实例没读到 env(改完 .env 没 restart 实例);有但仍异常看是否伴随 publish failed(redis 本身故障) | | 长任务跑到一半停下、提示「已达单轮步数上限…回复『继续』可接着跑」 | **预期行为非崩溃**:单个 run 自主步数到 backstop(`config/models/*.yaml` 的 `max_iterations`,flash 120 / pro 150)就主动停,回 `[reached max iterations]`。直接回复「继续」即续跑。经常撞顶=任务确实大,可调高对应 variant 的 `max_iterations` 或换 pro | | 任务停下提示「连续 N 步无净产出…已自动停止」(回 `[stopped: no progress]`) | **空转熔断**:连续 `_STALL_LIMIT`(loop.py,默认 8)步所有 tool 都只返 `[Error]`/重复结果/被拦 = 没在推进,主动停以免空烧 token。说明模型卡死在某个错上——**换思路 / 补充信息再回复「继续」**,别原样重发(会再次撞同一墙) | diff --git a/core/__init__.py b/core/__init__.py index 1b4c1f2..ea82a35 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,3 +1,3 @@ # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。 # 改版本只动这一行。 -__version__ = "0.41.2" +__version__ = "0.42.0" diff --git a/requirements.txt b/requirements.txt index 08923db..288406c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -27,6 +27,11 @@ croniter>=2.0 segno>=1.6 cryptography>=42.0 +# broker 外置(§7.0,蓝绿双实例跨进程 SSE/cancel):ZCBOT_REDIS_URL 设了才用, +# 不设走进程内 broker(dev 不需要起 redis);fakeredis 供单测(无需真 redis 服务) +redis>=5.0 +fakeredis>=2.23 + # §7 B 阶段: Storage 落 PG sqlalchemy>=2.0.0 psycopg[binary]>=3.1.0 diff --git a/tests/test_broker_redis.py b/tests/test_broker_redis.py new file mode 100644 index 0000000..6f3aa34 --- /dev/null +++ b/tests/test_broker_redis.py @@ -0,0 +1,177 @@ +"""RedisRunBroker 单测(fakeredis,无需真 redis 服务)。 + +覆盖跨进程语义的关键面: +- publish → 订阅 queue 收到(经 pubsub reader 路由) +- close 的 done key + publish 双通道:晚到订阅者(错过 publish)靠 key 立即收 done +- start 清 done key(新 run 的订阅者不被上一轮 done 打断) +- cancel key 的 request/is/clear 三态(跨"实例"可见:第二个 broker 实例读得到) +- emit 在 redis 挂掉时降级不抛(丢帧,run 不受影响) + +用两个 broker 实例共享同一 FakeServer 模拟蓝绿两进程。 +""" +import asyncio +import unittest +from uuid import uuid4 + +import fakeredis +import fakeredis.aioredis + +from web.broker import LocalRunBroker, RedisRunBroker, _make_broker + + +def _make_pair(server): + """一个共享 FakeServer 的 RedisRunBroker(sync + async 两个客户端注入)。""" + sync_c = fakeredis.FakeRedis(server=server, decode_responses=True) + async_c = fakeredis.aioredis.FakeRedis(server=server, decode_responses=True) + return RedisRunBroker("redis://fake:6379/0", sync_client=sync_c, async_client=async_c) + + +async def _drain_until(q: asyncio.Queue, timeout: float = 2.0): + return await asyncio.wait_for(q.get(), timeout=timeout) + + +class TestRedisBroker(unittest.TestCase): + def _run(self, coro): + return asyncio.run(coro) + + def test_emit_reaches_subscriber(self): + async def main(): + server = fakeredis.FakeServer() + b = _make_pair(server) + b.bind_loop(asyncio.get_running_loop()) + tid = uuid4() + q = b.subscribe(tid) + await asyncio.sleep(0.1) # 等 channel 挂载 + b.emit(tid, {"type": "text", "delta": "你好"}) + ev = await _drain_until(q) + self.assertEqual(ev, {"type": "text", "delta": "你好"}) + b.unsubscribe(tid, q) + await b.shutdown() + self._run(main()) + + def test_cross_instance_emit(self): + """模拟蓝绿:A 实例上的 run emit,B 实例上的 SSE 订阅者收到。""" + async def main(): + server = fakeredis.FakeServer() + b_a = _make_pair(server) # 跑 run 的实例 + b_b = _make_pair(server) # 挂 SSE 的实例 + loop = asyncio.get_running_loop() + b_a.bind_loop(loop) + b_b.bind_loop(loop) + tid = uuid4() + q = b_b.subscribe(tid) + await asyncio.sleep(0.1) + b_a.emit(tid, {"type": "llm_start"}) + ev = await _drain_until(q) + self.assertEqual(ev["type"], "llm_start") + await b_a.shutdown() + await b_b.shutdown() + self._run(main()) + + def test_late_subscriber_gets_done_via_key(self): + """close 之后才来的订阅者:publish 早已错过,靠 done key 立即断流。""" + async def main(): + server = fakeredis.FakeServer() + b = _make_pair(server) + b.bind_loop(asyncio.get_running_loop()) + tid = uuid4() + b.close(tid) # SETEX done + publish(无人订阅,publish 丢) + q = b.subscribe(tid) # 晚到 + ev = await _drain_until(q) + self.assertEqual(ev, {"type": "done"}) + self.assertTrue(b.is_done(tid)) + await b.shutdown() + self._run(main()) + + def test_start_clears_done(self): + async def main(): + server = fakeredis.FakeServer() + b = _make_pair(server) + b.bind_loop(asyncio.get_running_loop()) + tid = uuid4() + b.close(tid) + self.assertTrue(b.is_done(tid)) + b.start(tid) # 新 run + self.assertFalse(b.is_done(tid)) + q = b.subscribe(tid) + await asyncio.sleep(0.1) + self.assertTrue(q.empty()) # 不该被上一轮 done 打断 + b.emit(tid, {"type": "run_start"}) + ev = await _drain_until(q) + self.assertEqual(ev["type"], "run_start") + await b.shutdown() + self._run(main()) + + def test_cancel_cross_instance(self): + """A 实例(SSE/cancel 端点)request_cancel,B 实例(跑 run 的)poll 到。""" + async def main(): + server = fakeredis.FakeServer() + b_a = _make_pair(server) + b_b = _make_pair(server) + b_a.bind_loop(asyncio.get_running_loop()) + b_b.bind_loop(asyncio.get_running_loop()) + tid = uuid4() + self.assertFalse(b_b.is_cancelled(tid)) + b_a.request_cancel(tid) + self.assertTrue(b_b.is_cancelled(tid)) # 跨实例可见 + b_b.clear_cancel(tid) + self.assertFalse(b_a.is_cancelled(tid)) + await b_a.shutdown() + await b_b.shutdown() + self._run(main()) + + def test_emit_degrades_when_redis_down(self): + """redis 操作抛异常 → emit/is_cancelled 降级不抛(log 节流另测无必要)。""" + class Boom: + def __getattr__(self, name): + def _raise(*a, **k): + raise ConnectionError("redis down") + return _raise + + b = RedisRunBroker("redis://fake:6379/0", sync_client=Boom(), + async_client=fakeredis.aioredis.FakeRedis(decode_responses=True)) + tid = uuid4() + b.emit(tid, {"type": "text", "delta": "x"}) # 不抛 + self.assertFalse(b.is_cancelled(tid)) # 降级 False + b.request_cancel(tid) # 不抛 + b.close(tid) # 不抛 + + def test_bind_loop_fail_fast_when_unreachable(self): + class Boom: + def ping(self): + raise ConnectionError("refused") + + b = RedisRunBroker("redis://fake:6379/0", sync_client=Boom(), + async_client=fakeredis.aioredis.FakeRedis(decode_responses=True)) + + async def main(): + with self.assertRaises(RuntimeError): + b.bind_loop(asyncio.get_running_loop()) + self._run(main()) + + +class TestFactory(unittest.TestCase): + def test_default_is_local(self): + import os + old = os.environ.pop("ZCBOT_REDIS_URL", None) + try: + self.assertIsInstance(_make_broker(), LocalRunBroker) + finally: + if old is not None: + os.environ["ZCBOT_REDIS_URL"] = old + + def test_env_selects_redis(self): + import os + old = os.environ.get("ZCBOT_REDIS_URL") + os.environ["ZCBOT_REDIS_URL"] = "redis://127.0.0.1:6379/0" + try: + self.assertIsInstance(_make_broker(), RedisRunBroker) + finally: + if old is None: + del os.environ["ZCBOT_REDIS_URL"] + else: + os.environ["ZCBOT_REDIS_URL"] = old + + +if __name__ == "__main__": + unittest.main() diff --git a/web/app.py b/web/app.py index 1e4adf0..d4d47f9 100644 --- a/web/app.py +++ b/web/app.py @@ -1257,6 +1257,13 @@ def create_app() -> FastAPI: except Exception as e: print(f"[shutdown] sandbox shutdown_all error: {type(e).__name__}: {e}") + # broker 收尾(redis 版停 pubsub reader + 断连;local no-op)。放在 drain + # 之后 —— drain 期间 run 还要 emit/close。 + try: + await broker.shutdown() + except Exception as e: + print(f"[shutdown] broker shutdown error: {type(e).__name__}: {e}") + # drain 已 await inflight 收尾、run 线程退完;非阻塞关池(进程在退出,保守清理) run_executor.shutdown(wait=False) diff --git a/web/broker.py b/web/broker.py index 03db605..2defb65 100644 --- a/web/broker.py +++ b/web/broker.py @@ -1,38 +1,50 @@ -"""RunBroker:in-process pub/sub,把 agent run 产生的 event fan-out 给所有 SSE 订阅者。 +"""RunBroker:pub/sub,把 agent run 产生的 event fan-out 给所有 SSE 订阅者。 DESIGN §7 简化(0004 一并)—— 单活 run 形态下 run_id 是冗余的(同 task 同时 最多 1 个活 run);broker 内部全用 task_id 索引,客户端只需要 task_id 即可 订阅 / cancel,不再需要先拿 run_id。 -设计: -- emit() 从工作线程调(agent.run 在 to_thread 跑),用 loop.call_soon_threadsafe - 桥到 asyncio queue;SSE generator await queue.get() 拉出来推流。 -- 同一 task 多个订阅者(刷新页面 / 多 tab / 桌面+移动)— 每个订阅 1 个独立 queue。 -- run 结束 → broker.close(task_id) 给所有订阅者派一条 done;新订阅者(在 done - 后到的)立即收到 done 并断流(不漏不挂)。 -- 同 task 起新 run → broker.start(task_id) 清掉 _done 标记;否则上一轮 done - 会让新订阅者立刻断流看不到流式。 -- 进程内单实例 / 多进程不共享 — 个人 SaaS 单 worker 够用;真要扩多 worker 再上 Redis。 -- 不持久化 event — messages 已落 PG,刷新页面走 GET /v1/tasks/{id}/messages 看历史; - 真要"刷新继续看实时流"未来加 event log 表 + backfill。 +两个实现,env `ZCBOT_REDIS_URL` 决定(import 时定,进程生命周期不变): +- **LocalRunBroker**(默认,env 不设):进程内 asyncio queue,零依赖 —— 本地 dev / + 单实例部署用。多进程不共享:蓝绿并存期跨实例订阅/cancel 送达不到(RUN.md B 档边缘)。 +- **RedisRunBroker**(env 设了):event 走 Redis PUB/SUB(`zcbot:ev:`), + done/cancel 走带 TTL 的 key —— 蓝绿双实例跨进程可达,也是稳态双实例分流的前提 + (DESIGN §7.0)。启动 ping 不通 fail-fast(同 sandbox init 纪律);运行中 publish + 失败 log + 丢帧(直播降级,run 不受影响)。 -线程模型: -- broker.bind_loop(loop) 在 FastAPI startup 调一次,记录 asyncio loop 引用。 -- emit() 调用方可能在任意线程;put_nowait 是 thread-unsafe(asyncio.Queue 设计前提 - 是单 loop),所以走 call_soon_threadsafe 跨回 loop 线程再 put。 -- subscribe / unsubscribe / close / start 也都用 call_soon_threadsafe 包,避免 race - (实测 SSE generator 在 finally 里 unsubscribe,这个就在 loop 线程,直接调也行)。 +公共语义(两实现一致): +- emit() 从工作线程调(agent.run 在 to_thread 跑);SSE generator await queue.get()。 +- 同一 task 多个订阅者(刷新页面 / 多 tab)— 每个订阅 1 个独立 queue。 +- run 结束 → close(task_id) 给所有订阅者派 done;晚到的订阅者立即收 done 断流 + (local 用 _done set;redis 用 `zcbot:done:` 60s TTL key —— 主保护是 SSE + 端点先查 DB run_status,key 只兜"查完 DB 到订阅完成之间 run 恰好收尾"的窗口)。 +- 同 task 起新 run → start(task_id) 清 done 标记。 +- cancel:request_cancel 置位,AgentLoop 在 chunk 间 poll is_cancelled() 即退; + redis 版是 `zcbot:cancel:` key(GET ~0.1ms/次,localhost 可忽略),跨实例可达。 +- 不持久化 event — messages 已落 PG,刷新走 GET /messages 看历史;订阅建立前 + emit 的帧两个实现都会丢(设计选择)。 """ from __future__ import annotations import asyncio +import json +import os import threading +import time from collections import defaultdict from typing import Any, Optional from uuid import UUID -class RunBroker: +class LocalRunBroker: + """进程内实现(原 RunBroker)。 + + 线程模型: + - bind_loop(loop) 在 FastAPI startup 调一次,记录 asyncio loop 引用。 + - emit() 调用方可能在任意线程;put_nowait 是 thread-unsafe(asyncio.Queue 设计 + 前提是单 loop),所以走 call_soon_threadsafe 跨回 loop 线程再 put。 + """ + def __init__(self) -> None: self._subs: dict[UUID, set[asyncio.Queue]] = defaultdict(set) # 已经发完 done 的 task — 后来订阅者直接收到 done,避免无限等 @@ -47,6 +59,9 @@ class RunBroker: """FastAPI startup 调一次。""" self._loop = loop + async def shutdown(self) -> None: + """lifespan finally 调;local 无外部资源,no-op。""" + def start(self, task_id: UUID) -> None: """同 task 起新 run 时调:清 _done 标记,让新订阅者能看到流式。 cancel flag 在 finally 里 clear_cancel 清,这里不动(避免擦掉刚刚 request_cancel 的请求)。 @@ -98,7 +113,7 @@ class RunBroker: return len(self._subs.get(task_id, set())) def total_subscribers(self) -> int: - """全局 SSE 订阅者总数(所有 task 累加),供 §8.4 并发监控用。""" + """本进程 SSE 订阅者总数(所有 task 累加),供 §8.4 并发监控用。""" return sum(len(s) for s in self._subs.values()) def is_done(self, task_id: UUID) -> bool: @@ -121,5 +136,243 @@ class RunBroker: self._cancel_flags.pop(task_id, None) +class RedisRunBroker: + """Redis pub/sub 实现(蓝绿双实例跨进程,DESIGN §7.0)。 + + - event:`PUBLISH zcbot:ev:`(JSON);单条 async pubsub reader 任务 + 把消息路由到本进程各订阅 queue(订阅表 _subs 仍在本地 —— fan-out 的"最后 + 一公里"还是进程内,Redis 只负责跨进程那一跳)。 + - done:close 先 `SETEX zcbot:done: 60`再 publish done;subscribe 先挂 + channel 再查 key,晚到订阅者不漏。 + - cancel:`SETEX zcbot:cancel: 3600` / GET / DEL,跨实例可达。 + - 容错:bind_loop ping 不通直接 raise(fail-fast);运行中 redis 操作失败 + log(10s 节流)+ 降级(emit 丢帧、is_cancelled 返 False),run 不受影响; + reader 断线 1s 退避重连并重挂全部活跃 channel。 + """ + + _EV_PREFIX = "zcbot:ev:" + _DONE_PREFIX = "zcbot:done:" + _CANCEL_PREFIX = "zcbot:cancel:" + _CTL_CHANNEL = "zcbot:ev:__ctl__" # 常挂空 channel,保 pubsub 始终有订阅可 get_message + _DONE_TTL = 60 + _CANCEL_TTL = 3600 + + def __init__(self, url: str, *, sync_client=None, async_client=None) -> None: + import redis as _redis + import redis.asyncio as _aredis + self._url = url + # sync client:emit/close/cancel 从工作线程调(redis-py 自带连接池,线程安全); + # start/request_cancel 也会在 loop 线程上调 —— localhost 单次 ~0.1ms,超时收紧 + # 到 2s 防 redis 挂掉时把 event loop 卡死太久。 + self._redis = sync_client if sync_client is not None else _redis.Redis.from_url( + url, decode_responses=True, socket_timeout=2.0, socket_connect_timeout=2.0, + ) + self._aredis = async_client if async_client is not None else _aredis.from_url( + url, decode_responses=True, + ) + self._pubsub = None + self._reader_task: Optional[asyncio.Task] = None + self._subs: dict[UUID, set[asyncio.Queue]] = defaultdict(set) + self._loop: Optional[asyncio.AbstractEventLoop] = None + self._last_err_log = 0.0 + + # ─────────────── helpers ─────────────── + + def _chan(self, task_id: UUID) -> str: + return f"{self._EV_PREFIX}{task_id}" + + def _log_err(self, what: str, e: Exception) -> None: + """redis 故障时 emit/is_cancelled 每 chunk 都会撞,10s 节流防刷屏。""" + now = time.monotonic() + if now - self._last_err_log >= 10: + self._last_err_log = now + print(f"[broker] redis {what} failed: {type(e).__name__}: {e} (降级继续,10s 内同类不再报)") + + # ─────────────── lifecycle ─────────────── + + def bind_loop(self, loop: asyncio.AbstractEventLoop) -> None: + """FastAPI startup 调一次:ping 验活(不通 fail-fast)+ 起 pubsub reader。""" + self._loop = loop + try: + self._redis.ping() + except Exception as e: + raise RuntimeError( + f"ZCBOT_REDIS_URL 已设但 Redis 连不上({type(e).__name__}: {e})——" + f"确认 redis-server 在跑 / URL 对,或删掉该 env 回退进程内 broker" + ) + self._reader_task = loop.create_task(self._reader(), name="redis-broker-reader") + host = self._url.rsplit("@", 1)[-1] # 不把密码打进日志 + print(f"[broker] redis pub/sub enabled ({host})") + + async def shutdown(self) -> None: + """lifespan finally 调:停 reader、断两个客户端。""" + if self._reader_task is not None: + self._reader_task.cancel() + try: + await self._reader_task + except (asyncio.CancelledError, Exception): + pass + if self._pubsub is not None: + try: + await self._pubsub.aclose() + except Exception: + pass + try: + await self._aredis.aclose() + except Exception: + pass + try: + self._redis.close() + except Exception: + pass + + async def _reader(self) -> None: + """常驻:单条 pubsub 连接收所有活跃 channel 的消息,路由到本地订阅 queue。 + 断线 1s 退避重连,并把 _subs 里所有活跃 channel 重挂上(订阅状态不丢)。""" + while True: + try: + self._pubsub = self._aredis.pubsub() + channels = [self._CTL_CHANNEL] + [self._chan(t) for t in self._subs] + await self._pubsub.subscribe(*channels) + while True: + msg = await self._pubsub.get_message( + ignore_subscribe_messages=True, timeout=5.0 + ) + if msg is None or msg.get("type") != "message": + continue + ch = msg.get("channel") or "" + if ch == self._CTL_CHANNEL: + continue + try: + tid = UUID(ch[len(self._EV_PREFIX):]) + event = json.loads(msg["data"]) + except (ValueError, TypeError): + continue # 不认识的 channel / 坏 payload,丢 + for q in list(self._subs.get(tid, ())): + q.put_nowait(event) + except asyncio.CancelledError: + raise + except Exception as e: + self._log_err("reader", e) + try: + if self._pubsub is not None: + await self._pubsub.aclose() + except Exception: + pass + await asyncio.sleep(1) + + # ─────────────── pub/sub ─────────────── + + def start(self, task_id: UUID) -> None: + try: + self._redis.delete(f"{self._DONE_PREFIX}{task_id}") + except Exception as e: + self._log_err("start/del-done", e) + + def subscribe(self, task_id: UUID) -> asyncio.Queue: + """调用方:SSE handler(loop 线程)。queue 立即返回;channel 挂载与 done-key + 补查异步完成 —— 挂载完成前 publish 的帧会丢(与 local 版"订阅前的帧丢"同类, + SSE 端点已先查过 DB run_status 兜大头)。""" + q: asyncio.Queue = asyncio.Queue() + first = not self._subs[task_id] + self._subs[task_id].add(q) + + async def _setup() -> None: + try: + if first and self._pubsub is not None: + await self._pubsub.subscribe(self._chan(task_id)) + # 先挂 channel 再查 done key:close() 是先 SETEX 再 publish, + # 两边一交叉,晚到订阅者要么收到 publish、要么查到 key,不漏 + if await self._aredis.exists(f"{self._DONE_PREFIX}{task_id}"): + q.put_nowait({"type": "done"}) + except Exception as e: + self._log_err("subscribe-setup", e) + q.put_nowait({"type": "done"}) # 保守断流,客户端自动重连 + + if self._loop is not None: + self._loop.create_task(_setup()) + return q + + def unsubscribe(self, task_id: UUID, q: asyncio.Queue) -> None: + subs = self._subs.get(task_id) + if subs is None: + return + subs.discard(q) + if not subs: + del self._subs[task_id] + ch = self._chan(task_id) + + async def _unsub() -> None: + try: + if self._pubsub is not None: + await self._pubsub.unsubscribe(ch) + except Exception: + pass # reader 重连时按 _subs 重挂,残留订阅无害 + + if self._loop is not None: + self._loop.create_task(_unsub()) + + def emit(self, task_id: UUID, event: dict[str, Any]) -> None: + try: + self._redis.publish( + self._chan(task_id), json.dumps(event, ensure_ascii=False, default=str) + ) + except Exception as e: + self._log_err("publish", e) # 丢帧降级,run 不受影响 + + def close(self, task_id: UUID) -> None: + try: + self._redis.set(f"{self._DONE_PREFIX}{task_id}", "1", ex=self._DONE_TTL) + except Exception as e: + self._log_err("close/set-done", e) + self.emit(task_id, {"type": "done"}) + + def n_subscribers(self, task_id: UUID) -> int: + return len(self._subs.get(task_id, set())) + + def total_subscribers(self) -> int: + """本进程的订阅数(蓝绿双实例各报各的,admin 面板看单实例)。""" + return sum(len(s) for s in self._subs.values()) + + def is_done(self, task_id: UUID) -> bool: + try: + return bool(self._redis.exists(f"{self._DONE_PREFIX}{task_id}")) + except Exception as e: + self._log_err("is_done", e) + return False + + # ─────────────── cancel signaling ─────────────── + + def request_cancel(self, task_id: UUID) -> None: + try: + self._redis.set(f"{self._CANCEL_PREFIX}{task_id}", "1", ex=self._CANCEL_TTL) + except Exception as e: + self._log_err("request_cancel", e) + + def is_cancelled(self, task_id: UUID) -> bool: + """AgentLoop 在 stream chunk 间 poll(工作线程);redis 挂了返 False 降级 + (cancel 暂不可达,等价于 redis 前的跨实例行为,run 不受影响)。""" + try: + return self._redis.get(f"{self._CANCEL_PREFIX}{task_id}") is not None + except Exception as e: + self._log_err("is_cancelled", e) + return False + + def clear_cancel(self, task_id: UUID) -> None: + try: + self._redis.delete(f"{self._CANCEL_PREFIX}{task_id}") + except Exception as e: + self._log_err("clear_cancel", e) + + +def _make_broker(): + """env `ZCBOT_REDIS_URL` 在 → Redis 版(蓝绿跨进程);不在 → 进程内版(dev/单实例)。 + import 时定一次,与 ZCBOT_INSTANCE 同款读法。""" + url = os.getenv("ZCBOT_REDIS_URL", "").strip() + if url: + return RedisRunBroker(url) + return LocalRunBroker() + + # 进程内单例 — FastAPI lifespan 里 bind_loop;agent / sink / SSE handler 共享。 -broker = RunBroker() +broker = _make_broker() diff --git a/web/sinks.py b/web/sinks.py index 4c581c4..40c7b0a 100644 --- a/web/sinks.py +++ b/web/sinks.py @@ -9,11 +9,9 @@ from __future__ import annotations from typing import Any from uuid import UUID -from .broker import RunBroker - - class WebEventSink: - def __init__(self, broker: RunBroker, task_id: UUID) -> None: + def __init__(self, broker, task_id: UUID) -> None: + # broker: LocalRunBroker | RedisRunBroker(鸭子类型同接口,web/broker.py) self._broker = broker self._task_id = task_id