From e52d6880f577ee8fea5f76d28e97f5258eed5ff8 Mon Sep 17 00:00:00 2001 From: Junyan Qin Date: Wed, 29 Jul 2026 12:47:53 +0800 Subject: [PATCH] fix(cloud): eliminate periodic runtime CPU spikes --- docs/multi-tenant/cloud-runtime-soak-gate.md | 4 +- .../cloud-v2-pending-verification.md | 23 ++++- .../runtime-resource-audit-2026-07-28.md | 40 +++++---- pyproject.toml | 2 +- scripts/cloud_runtime_soak.py | 5 ++ src/langbot/pkg/cloud/directory_projection.py | 40 +++++++++ src/langbot/pkg/core/app.py | 20 +++++ src/langbot/pkg/pipeline/aggregator.py | 25 ++++-- src/langbot/pkg/provider/tools/loaders/mcp.py | 84 ++++++++++++++----- .../cloud/test_directory_projection.py | 24 ++++++ tests/unit_tests/pipeline/test_aggregator.py | 51 +++++++++++ .../unit_tests/provider/test_mcp_resources.py | 81 ++++++++++++++---- .../utils/test_cloud_runtime_soak.py | 35 ++++++++ uv.lock | 4 +- 14 files changed, 372 insertions(+), 66 deletions(-) diff --git a/docs/multi-tenant/cloud-runtime-soak-gate.md b/docs/multi-tenant/cloud-runtime-soak-gate.md index 33d6f75cb..7118ffa07 100644 --- a/docs/multi-tenant/cloud-runtime-soak-gate.md +++ b/docs/multi-tenant/cloud-runtime-soak-gate.md @@ -89,7 +89,9 @@ uv run python scripts/cloud_runtime_soak.py \ - event-loop recent p95 不超过 `--max-event-loop-p95-lag-ms`。 - blocking executor `pending` 至少回到过零；不能整个尾段持续积压。 - Plugin Runtime restart coordinator 的 active launch、half-open probe 和 - circuit open remaining time 必须回到零。 + circuit open remaining time 必须回到零，`gate_waiters` 必须至少归零一次。 +- Core 的 MCP projection retirement queue/worker 和 message aggregation + buffer/scope 必须至少归零一次。 - telemetry、QueryPool、MCP host/dispatch、Box creating/closing/background 等临时 gauge 不能继续增长。 内存判定要求“增长量”和“斜率”同时越界，避免几 MiB allocator/page-cache 噪声在短窗口被外推成很大的每小时斜率。最终报告仍保留实际增长与斜率，人工审查时不能只看 verdict。 diff --git a/docs/multi-tenant/cloud-v2-pending-verification.md b/docs/multi-tenant/cloud-v2-pending-verification.md index 283fd0771..df768a301 100644 --- a/docs/multi-tenant/cloud-v2-pending-verification.md +++ b/docs/multi-tenant/cloud-v2-pending-verification.md @@ -16,7 +16,8 @@ ## 1. 当前已形成的交付基线 -- LangBot Core 和 Plugin SDK 的全量测试、Ruff、`git diff --check` 已通过。 +- LangBot Core 全量 `2811 passed, 33 skipped`，Plugin SDK 全量 + `1317 passed`；两仓 Ruff、`git diff --check` 已通过。 - Plugin Runtime 和 Box Runtime 的公开健康接口、event-loop lag 与有界 blocking executor 指标已经过真实进程短时验证。 - 仓库 Dockerfile 构建的 Linux/cgroup v2 短时探针已证明 CPU、memory、 @@ -24,8 +25,12 @@ - PostgreSQL 16 + RLS 的 1,000 Workspace 真实启动测试，以及 5,000 Workspace 三代替换合成探针已通过。 - Core 已精确钉住 Plugin SDK 提交 - `a5a96b302a5808af84bbdd28833ce050176f87e9`。最终验证必须使用包含该提交的 + `c7893f8f94e83cb09ebcc98e25490fa699b0fb69`。最终验证必须使用包含该提交的 Core、Plugin Runtime 和 Box Runtime 镜像，不能混用旧 SDK。 +- 独立资源复核已经移除 Cloud MCP 每会话 5 秒查询执行绑定的轮询，改由签名目录 + 投影提交后向一个合并回收任务发布代次变化；工具与资源调用前后仍使用数据库 + execution fence。Plugin restart 冷却等待者、MCP 投影回收、消息聚合 buffer/scope + 均已纳入健康快照和 soak 归零门禁。 以上结果是进入生产候选验证的前提，不是 SaaS 上线批准。 @@ -126,7 +131,13 @@ 一个副本追平不能使另一个副本跳过本地 cache 刷新。 7. 同时使大量 plugin worker 因系统性故障退出，证明 restart launch 全局并发受限、 失败阈值触发 Runtime circuit、冷却后只有一个 half-open probe，且 probe 未稳定前 - 其他 installation 不会继续重启；24 小时门禁必须把 circuit 打开判为失败。 + 其他 installation 不会继续重启；冷却计时器/状态等待者数量不得超过全局 restart + 并发，取消 probe 不得把 circuit 永久卡在 half-open；24 小时门禁必须把 circuit + 打开或 `gate_waiters` 未归零判为失败。 +8. 使用大量空闲 remote MCP session 做 generation 切换，证明目录投影只创建一个 + 合并回收任务，不产生每 session 周期数据库查询、计时器或同时唤醒；旧 session + 最终关闭，`mcp_projection_retirements` 和 + `mcp_projection_reconcile_active` 在冷却期归零。 ### V-06：套餐、Box 与 stdio MCP @@ -160,6 +171,8 @@ PostgreSQL/pgvector 和代表性 Workspace 配置分布，测量： - 空 Workspace、活跃 Workspace、每个启用插件和每个 Pro sandbox 的边际 RSS、线程、文件描述符、连接和 PostgreSQL pool 成本； - 启动、目录重放、批量 reconcile 和故障恢复的耗时与峰值； +- remote MCP 数量增加及目录 generation 批量切换时的数据库 QPS、回收队列和 + event-loop lag，确认不存在与 session 数量成比例的空闲轮询； - 单实例可批准的 Workspace、活跃 Bot、plugin worker 和 sandbox 上限。 容量上限必须写入生产配置与告警，不能只保留在测试报告中。 @@ -173,13 +186,15 @@ PostgreSQL/pgvector 和代表性 Workspace 配置分布，测量： 2. plugin reconcile、依赖准备、调用、崩溃与重启； 3. Dashboard/Embed/平台 WebSocket 建连、突发消息和断连；HTTP Bot 覆盖 高基数 session/idempotency、硬容量拒绝、空闲回收及 callback 堵塞； + remote MCP 覆盖大量空闲连接、批量 generation 切换和合并回收； 4. Box session、文件同步、并发 exec、输出与清理； 5. PostgreSQL pool 接近容量、事务超时和恢复； 6. Core、Plugin Runtime、Box 分别 SIGTERM 和恢复。 最后至少保留 30 分钟无测试流量冷却。任一健康失败、OOM/memory pressure、 PID limit、blocking executor rejection、超阈值 CPU throttling/event-loop lag、 -冷却尾段内存持续增长或临时 gauge 不回落都判为失败。 +冷却尾段内存持续增长，或 Plugin restart `gate_waiters`、MCP 投影回收、 +消息聚合 buffer/scope 等临时 gauge 不回落都判为失败。 必须归档： diff --git a/docs/multi-tenant/runtime-resource-audit-2026-07-28.md b/docs/multi-tenant/runtime-resource-audit-2026-07-28.md index 6a31111c4..ba25e00ef 100644 --- a/docs/multi-tenant/runtime-resource-audit-2026-07-28.md +++ b/docs/multi-tenant/runtime-resource-audit-2026-07-28.md @@ -17,7 +17,7 @@ 1. 在最终 Cloud 部署权限和 cgroup 拓扑下重复 nsjail、namespace 和 delegated cgroup v2 的 CPU、内存、swap、PID、文件句柄验证。本轮一次性 Linux 容器已经证明代码路径可工作，但普通容器和仅 `--privileged` 的 private cgroup namespace 都不满足条件。 2. 为 Cloud Box 提供并验证硬文件系统 quota provider。普通 nsjail bind mount 不能证明总字节数和 inode 硬配额，当前严格 readiness 按设计会失败关闭。 3. 使用最终生产配置分布继续做容量测试，并据此确定单实例 Workspace placement 上限。本轮真实 PostgreSQL 16 + RLS 启动测试已经覆盖 1,000 个各带 Provider、三类 Model、Bot、Pipeline、KnowledgeBase、MCP 和 Plugin setting 的 Workspace，启动加载耗时和 SQL 次数保持线性；5,000 Workspace 的合成三代替换探针也证明旧运行时会释放。仓库已新增可同时采集 Core/Plugin/Box HTTP、进程树和 cgroup v2 的 24 小时门禁工具，并在受 CPU、memory、swap、PID 硬限制的 Linux 容器中完成短时自检；但最终生产候选拓扑的 24 小时运行仍未执行。测试中的 fake adapter/requester/Plugin handler 仍不能替代真实平台 SDK、外部连接池和插件进程的容量数据；合法活跃租户本身仍会线性占用内存。 -SDK 已先行发布到分支提交 `a5a96b302a5808af84bbdd28833ce050176f87e9`，本提交集中的 LangBot +SDK 已先行发布到分支提交 `c7893f8f94e83cb09ebcc98e25490fa699b0fb69`，本提交集中的 LangBot `pyproject.toml` 和 `uv.lock` 已精确钉住该提交。最终镜像仍需按待验证清单记录并核对实际安装版本。 ## 覆盖范围 @@ -70,6 +70,18 @@ SDK 已先行发布到分支提交 `a5a96b302a5808af84bbdd28833ce050176f87e9`， - Session、Conversation、WebSocket connection/proxy/message、rate-limit identity、task record/log、telemetry task、vector handle 和 adapter 私有队列均有容量或 LRU/TTL。 - SessionManager 现在维护 Workspace 二级索引和带 revision 校验的最小过期堆。新会话只扫描目标 Workspace 的有界会话集，TTL 回收只消费已过期堆前缀，全局 idle 淘汰使用最小堆；高频命中产生的旧堆项按活跃会话的有界倍数压缩。原实现会在每个攻击者可制造的新 launcher id 上扫描并排序实例全部会话。 - SDK 的 EventContext 和依赖准备锁使用 weak reference；generation、admission、installation、capability 和 completed-process 状态有上限。 +- Plugin restart circuit 打开期间只有 `max_concurrent_restarts` 个 supervisor + 能持有冷却计时器/状态等待，其他 installation 睡眠在同一个 semaphore FIFO； + probe 状态变更在调用者取消时仍会完成，避免 half-open 永久占用。 +- Box nsjail 启动时只扫描一次 `/proc`，并流式删除遗留 session 目录；不再为每个 + 遗留目录重复扫描全部进程或先把全部目录物化到内存。 +- Box Skill discovery、目录列表和列表正文分别限制扫描 entry、package、返回 entry + 与累计文本字节；BFS 使用 deque，拒绝 inode 洪泛导致的 O(N²) 或无界列表。 +- Core message aggregation 使用 `(instance, workspace, generation)` O(1) scope + counter 做准入，不再为每个新 launcher 扫描全实例 buffer。 +- Cloud remote MCP 的 idle execution fence 不再由每个 session 每 5 秒查询 + Workspace/ExecutionState；签名目录投影事务提交后将失效 scope 合并进一个有界 + cleanup worker。实际工具/资源调用前后仍保留数据库强校验。 - 空 Workspace 不再预分配 Model generation scope、Plugin installation set 或 Box generation event；只有 Workspace 实际拥有对应运行时资源或等待任务时才创建这些对象。 - Runtime RPC 文件同时限制单文件字节数和单连接未消费文件数量，连接关闭时清理连接拥有的临时文件。 - Box Runtime 维护实例与 Workspace 到活跃 session 的二级索引；创建、删除、过期、撤销和 shutdown 共用同一清理路径，避免每个租户 RPC 都扫描实例中的全部 session。 @@ -159,20 +171,20 @@ SDK 已先行发布到分支提交 `a5a96b302a5808af84bbdd28833ce050176f87e9`， | --- | --- | | LangBot Ruff + `git diff --check` | 通过 | | Plugin SDK Ruff + `git diff --check` | 通过 | -| LangBot 全量测试（使用远端精确钉住的新 SDK，含 unit/integration/Box/E2E） | `2808 passed, 33 skipped` | -| Plugin SDK 全量测试 | `1312 passed` | +| LangBot 全量测试（使用远端精确钉住的新 SDK，含 unit/integration/Box/E2E） | `2811 passed, 33 skipped` | +| Plugin SDK 全量测试 | `1317 passed` | | 真实 PostgreSQL 16 + pgvector 迁移/RLS/发布测试（严格资源告警） | `22 passed` | | 真实 PostgreSQL 16 + RLS populated Cloud 启动容量 | 500 Workspace `6.178s / CPU 3.026s`；当前 1,000 Workspace 复跑 `12.109s / CPU 5.967s` | | 较早 Core Dockerfile Linux 镜像构建与 `regex` 导入 | 通过，image SHA `8893a14053df`；该镜像使用旧 SDK pin，已失效，最终候选必须重建 | | `ResourceWarning` + `PytestUnraisableExceptionWarning` 全量门禁 | Core 与 SDK 均通过，并已固化到 pytest 配置 | | Plugin SDK Box 专项测试（含全局扫描回归保护） | `669 passed` | | Docker Compose 渲染、Compose/Kubernetes YAML 解析与 diff 检查 | 通过 | -| Cloud soak 门禁解析/采样/判定单元测试 | `24 passed` | +| Cloud soak 门禁解析/采样/判定单元测试 | `25 passed` | | Core/Plugin SDK event-loop monitor 专项测试 | 两仓各 `7 passed`，包含真实 50 ms scheduler stall | | Cloud soak Linux 硬限制短时自检 | 通过；CPU `0.5`、memory+swap `256 MiB`、PID `128` 均从 cgroup v2 读回，冷却尾段 verdict `pass` | -| Core 双阶段历史 churn 资源探针（使用本地新 SDK） | audit 通过，`11.275s` | -| Core 5,000 个 populated Workspace 三代容量探针（使用本地新 SDK） | audit 通过，最大替换耗时比 `1.378` | -| Plugin SDK 双阶段资源探针 | audit 通过，`9.037s` | +| Core 双阶段历史 churn 资源探针（使用精确钉住的新 SDK） | audit 通过，`11.895s` | +| Core 5,000 个 populated Workspace 三代容量探针（使用精确钉住的新 SDK） | audit 通过，最大替换耗时比 `1.435` | +| Plugin SDK 双阶段资源探针 | audit 通过，`9.407s` | 两个仓库新增了可重复执行的历史 churn 探针，Core 另有 populated Workspace 三代替换探针： @@ -202,24 +214,24 @@ Core audit 每个阶段执行 10,000 个空 Workspace 的真实 Model/Plugin man - 20,000 个限流身份：rate-limit container `10,000`。 - 10,000 个历史 task：task record `200`。 - 5,000 次 WebSocket churn：conversation 与 stream index 均为 `200`。 -- event-loop task、线程和文件描述符保持 `1 / 1 / 6`；强制加载本地最新 SDK 的当前复跑中，第二阶段相对第一阶段 RSS 增长 `1,605,632 bytes`、tracemalloc current 增长 `344,735 bytes`，总耗时 `11.275s`。Session 淘汰改为 Workspace 索引和最小堆后，同一 audit 工作量相对此前 `16.150s` 明显下降。 +- event-loop task、线程和文件描述符保持 `1 / 1 / 6`；使用远端精确钉住 SDK 的当前复跑中，第二阶段相对第一阶段 RSS 增长 `2,605,056 bytes`、tracemalloc current 增长 `344,622 bytes`，总耗时 `11.895s`。Session 淘汰改为 Workspace 索引和最小堆后，同一 audit 工作量相对此前 `16.150s` 明显下降。 Populated Workspace audit 为 5,000 个 Workspace 各加载一个 Provider、LLM、Embedding、Rerank、Pipeline、Bot、KnowledgeBase 和 MCP session，然后全部推进两个 generation： - 三个阶段的活跃 provider/model、pipeline、bot、knowledge 和 MCP registry 均精确维持 `5,000`，不存在按历史 generation 增长。 - 到第三阶段，前两代的 requester、Bot adapter 和 MCP session 各 `10,000` 个全部收到确定性关闭；weak reference 断言旧代对象可被回收。 -- event-loop task、线程和文件描述符保持 `1 / 1 / 6`；强制加载本地最新 SDK 的当前复跑中，第三阶段相对第二阶段 RSS 增长 `1,277,952 bytes`，tracemalloc current 反而减少 `148 bytes`。 -- 初始/第一次替换/第二次替换分别耗时 `1.707s / 2.205s / 2.351s`，最大替换耗时比为 `1.378`，未随历史代次出现 CPU 退化。 -- macOS RSS sample 从初始的 `154,271,744` 增至第一阶段 `367,820,800`、第二阶段 `388,726,784` 和第三阶段 `390,004,736 bytes`；第二次替换只比第一次替换增加约 1.22 MiB，但“合法活跃租户资源的线性容量”仍必须作为 placement 容量输入。这里使用轻量 fake adapter/requester，不应把第一阶段约 204 MiB 增量外推为生产每租户成本。 +- event-loop task、线程和文件描述符保持 `1 / 1 / 6`；使用远端精确钉住 SDK 的当前复跑中，第三阶段相对第二阶段 RSS 增长 `1,261,568 bytes`，tracemalloc current 仅增长 `510 bytes`。 +- 初始/第一次替换/第二次替换分别耗时 `1.804s / 2.420s / 2.588s`，最大替换耗时比为 `1.435`，未随历史代次出现 CPU 退化。 +- macOS RSS sample 从初始的 `154,484,736` 增至第一阶段 `368,050,176`、第二阶段 `388,939,776` 和第三阶段 `390,201,344 bytes`；第二次替换只比第一次替换增加约 1.20 MiB，但“合法活跃租户资源的线性容量”仍必须作为 placement 容量输入。这里使用轻量 fake adapter/requester，不应把第一阶段约 204 MiB 增量外推为生产每租户成本。 Plugin SDK audit 每个阶段执行 25,000 次 loopback RPC、5,000 次安装 binding 激活/撤销、10,000 个 Workspace generation 更新和 2,500 次带 Workspace 上下文的 Box session 创建/删除。第一、第二阶段的保留状态完全一致： - RPC waiter、stream queue、action task 和活跃 installation binding 均为 `0`。 - installation watermark 为有界的 `5,000`；Workspace generation record 为有界的 `10,000`，没有等待者时 generation event 为 `0`。 - generation active task/index、Box session、Box Workspace session index、creating/closing/background task 和 session lock 均为 `0`。 -- event-loop task 和文件描述符保持 `1 / 7`；当前复跑第二阶段相对第一阶段 RSS peak 增长 `2,654,208 bytes`、tracemalloc current 增长 `289,634 bytes`，总耗时 `9.037s`。耗时增加来自本轮把大协议消息的 JSON/Pydantic、UTF-8 编码、分片和拼接移入有界线程池；25,000 RPC/阶段的合成探针仍约为 5,500 RPC/s，结构状态和第二阶段 tracemalloc 增量保持平稳。 +- event-loop task 和文件描述符保持 `1 / 7`；当前复跑第二阶段相对第一阶段 RSS peak 增长 `2,670,592 bytes`、tracemalloc current 增长 `289,746 bytes`，总耗时 `9.407s`。耗时增加来自本轮把大协议消息的 JSON/Pydantic、UTF-8 编码、分片和拼接移入有界线程池；25,000 RPC/阶段的合成探针仍约为 5,300 RPC/s，结构状态和第二阶段 tracemalloc 增量保持平稳。 -第二轮反向静态审查另外枚举了 Core 的 50 个显式 task 创建点和 204 个线程、阻塞调用及子进程调用点，以及 SDK 的 28 个显式 task 创建点和 62 个线程、阻塞调用及子进程调用点。显式 task 均具有持有者、完成回调或 `finally` 回收路径；所有生产入口在第一次 `asyncio.to_thread()` 前安装有界默认 executor。Core、Plugin Runtime 和 Box 的公开 `/healthz`（Box `/readyz` 亦同）会输出各自的 aggregate runtime/resource counter 和 event-loop lag，供 soak 对比活跃量、pending、累计 capacity rejection 与调度延迟；不输出 debug key、控制 token、租户或插件身份。Plugin Runtime 的授权 debug info 复用同一资源快照，避免公开/私有指标语义漂移。 +第二轮反向静态审查另外枚举了 Core 的 50 个显式 task 创建点和 204 个线程、阻塞调用及子进程调用点，以及 SDK 的 28 个显式 task 创建点和 62 个线程、阻塞调用及子进程调用点。第三轮独立复核继续从高基数定时器、目录遍历、准入全表扫描和取消竞态反推，新增关闭了 Plugin restart 冷却唤醒群、MCP idle 数据库轮询、nsjail orphan 的 O(session × process) 启动扫描、message aggregation 的 O(buffer) 准入及 Skill inode/文本列表边界。显式 task 均具有持有者、完成回调或 `finally` 回收路径；所有生产入口在第一次 `asyncio.to_thread()` 前安装有界默认 executor。Core、Plugin Runtime 和 Box 的公开 `/healthz`（Box `/readyz` 亦同）会输出各自的 aggregate runtime/resource counter 和 event-loop lag，供 soak 对比活跃量、pending、累计 capacity rejection 与调度延迟；不输出 debug key、控制 token、租户或插件身份。Plugin Runtime 的授权 debug info 复用同一资源快照，避免公开/私有指标语义漂移。 真实 PostgreSQL populated 启动门禁会先通过 release migration 创建最新 schema，再用无 `BYPASSRLS` 的临时 Cloud Runtime 角色启动。每个 Workspace 都含九类代表性资源，测试会走实际的 instance discovery、tenant UoW、启动 binding 快照和 Model/Platform/Pipeline/RAG/MCP/Plugin 加载路径： @@ -230,7 +242,7 @@ Plugin SDK audit 每个阶段执行 25,000 次 loopback RPC、5,000 次安装 bi 探针要求第二阶段的结构状态与第一阶段精确相等，并对第二阶段 RSS 与 tracemalloc 增长设置失败阈值。macOS 的 RSS 来源是 `getrusage` peak，因此这里验证的是峰值增量边界而非“当前 RSS 回落”；最终 Linux 24 小时 soak 仍需采集 current RSS/PSS 和 cgroup `memory.current`。 -LangBot 全量测试的 33 个 skip 中，22 个是默认全量运行未提供 PostgreSQL/pgvector 而跳过的集成用例，10 个是未提供 Valkey，另 1 个是可选环境的 collection skip；真实 PostgreSQL 相关路径已由上表单独运行覆盖。Plugin SDK 的 22 个 warning 为现有 Pydantic v2 deprecation 与 aiohttp AppKey 建议；没有失败、未关闭资源或资源上限降级。Core 当前全量产生 192 个既有第三方/兼容性 warning；`ResourceWarning` 和 `PytestUnraisableExceptionWarning` 仍由 pytest 配置提升为错误，本轮没有此类泄漏告警。 +LangBot 全量测试的 33 个 skip 中，22 个是默认全量运行未提供 PostgreSQL/pgvector 而跳过的集成用例，10 个是未提供 Valkey，另 1 个是可选环境的 collection skip；真实 PostgreSQL 相关路径已由上表单独运行覆盖。Plugin SDK 的 26 个 warning 为现有 Pydantic v2 deprecation 与 aiohttp AppKey 建议；没有失败、未关闭资源或资源上限降级。Core 当前全量产生 193 个既有第三方/兼容性 warning；`ResourceWarning` 和 `PytestUnraisableExceptionWarning` 仍由 pytest 配置提升为错误，本轮没有此类泄漏告警。 Linux Runtime 探针使用上述镜像并只读挂载本地最新 SDK 源码： diff --git a/pyproject.toml b/pyproject.toml index 8061b385f..06015bd44 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -71,7 +71,7 @@ dependencies = [ "chromadb>=1.0.0,<2.0.0", "qdrant-client (>=1.15.1,<2.0.0)", "pyseekdb==1.1.0.post3", - "langbot-plugin @ git+https://github.com/langbot-app/langbot-plugin-sdk.git@a5a96b302a5808af84bbdd28833ce050176f87e9", + "langbot-plugin @ git+https://github.com/langbot-app/langbot-plugin-sdk.git@c7893f8f94e83cb09ebcc98e25490fa699b0fb69", "asyncpg>=0.30.0", "line-bot-sdk>=3.19.0", "matrix-nio>=0.25.2", diff --git a/scripts/cloud_runtime_soak.py b/scripts/cloud_runtime_soak.py index 8d16c60b1..f0eafdd89 100644 --- a/scripts/cloud_runtime_soak.py +++ b/scripts/cloud_runtime_soak.py @@ -76,8 +76,13 @@ RUNTIME_FAILURE_COUNTER_SUFFIXES = ('.restart_coordinator.circuit_open_total',) DRAIN_GAUGE_SUFFIXES = ( '.blocking_executor.pending', '.restart_coordinator.active_launches', + '.restart_coordinator.gate_waiters', '.restart_coordinator.half_open_probe_inflight', '.restart_coordinator.open_remaining_seconds', + '.resources.runtimes.mcp_projection_retirements', + '.resources.runtimes.mcp_projection_reconcile_active', + '.resources.runtimes.message_aggregation_buffers', + '.resources.runtimes.message_aggregation_scopes', ) TRANSIENT_GAUGE_SUFFIXES = ( '.resources.telemetry_tasks', diff --git a/src/langbot/pkg/cloud/directory_projection.py b/src/langbot/pkg/cloud/directory_projection.py index 0430ef07e..ded071ca6 100644 --- a/src/langbot/pkg/cloud/directory_projection.py +++ b/src/langbot/pkg/cloud/directory_projection.py @@ -259,6 +259,7 @@ class DirectoryProjectionService: await session.flush() await self._reconcile_entitlement_snapshot_set(snapshot) + self._publish_runtime_execution_projection(snapshot.workspaces) self._record_success() self._consumer_cursor = snapshot.cursor @@ -349,10 +350,49 @@ class DirectoryProjectionService: returned.values(), requested_workspace_uuids=requested, ) + self._publish_runtime_execution_projection( + returned.values(), + affected_workspace_uuids=requested, + ) if projection_caught_up: self._record_success() self._consumer_cursor = batch.cursor + def _publish_runtime_execution_projection( + self, + workspaces: Iterable[DirectoryWorkspace], + *, + affected_workspace_uuids: set[str] | None = None, + ) -> None: + """Retire stale runtime scopes without per-session database polling. + + The signed directory transaction is already committed when this hook + runs. Runtime calls still validate the database fence before and after + side effects; this notification only releases idle resources promptly. + """ + + tool_manager = getattr(self.ap, 'tool_mgr', None) + mcp_loader = getattr(tool_manager, 'mcp_tool_loader', None) + reconcile = getattr(mcp_loader, 'reconcile_execution_projection', None) + if not callable(reconcile): + return + active_generations = { + workspace.uuid: workspace.execution_generation + for workspace in workspaces + if workspace.status == WorkspaceStatus.ACTIVE.value + } + try: + reconcile( + self.instance_uuid, + active_generations, + affected_workspace_uuids=affected_workspace_uuids, + ) + except Exception: + # Runtime retirement is a resource cleanup path, not an execution + # admission boundary. Database-backed call-time fences remain + # authoritative if a local runtime hook fails. + self.ap.logger.exception('Failed to publish the Cloud execution projection to MCP runtimes') + async def _reconcile_entitlement_snapshot_set( self, snapshot: DirectorySnapshot, diff --git a/src/langbot/pkg/core/app.py b/src/langbot/pkg/core/app.py index fb461b056..de49de979 100644 --- a/src/langbot/pkg/core/app.py +++ b/src/langbot/pkg/core/app.py @@ -231,6 +231,14 @@ class Application: 'bots': len(getattr(self.platform_mgr, '_bots_by_key', {})), 'pipelines': len(getattr(self.pipeline_mgr, '_pipelines_by_key', {})), 'knowledge_bases': len(getattr(self.rag_mgr, 'knowledge_bases', {})), + 'message_aggregation_buffers': len(getattr(self.msg_aggregator, 'buffers', {})), + 'message_aggregation_scopes': len( + getattr( + self.msg_aggregator, + '_buffer_counts_by_scope', + {}, + ) + ), 'plugin_installations': len( getattr( self.plugin_connector, @@ -245,6 +253,18 @@ class Application: 'mcp_sessions': len(getattr(mcp_loader, '_sessions', {})), 'mcp_host_tasks': len(getattr(mcp_loader, '_hosted_mcp_tasks', ())), 'mcp_dispatch_tasks': len(getattr(mcp_loader, '_host_dispatch_tasks', ())), + 'mcp_projection_retirements': len(getattr(mcp_loader, '_pending_projection_retirements', ())), + 'mcp_projection_reconcile_active': int( + ( + projection_task := getattr( + mcp_loader, + '_projection_reconcile_task', + None, + ) + ) + is not None + and not projection_task.done() + ), } ) diff --git a/src/langbot/pkg/pipeline/aggregator.py b/src/langbot/pkg/pipeline/aggregator.py index de9b206ca..c24468306 100644 --- a/src/langbot/pkg/pipeline/aggregator.py +++ b/src/langbot/pkg/pipeline/aggregator.py @@ -71,6 +71,10 @@ class MessageAggregator: def __init__(self, ap: app.Application): self.ap = ap self.buffers = {} + self._buffer_counts_by_scope: dict[ + tuple[str, str, int], + int, + ] = {} self.lock = asyncio.Lock() concurrency = self.ap.instance_config.data.get('concurrency', {}) self.max_buffers = max(int(concurrency.get('pending_queries', 1000)), 1) @@ -194,12 +198,14 @@ class MessageAggregator: async with self.lock: buffer = self.buffers.get(aggregation_key) if buffer is None: - workspace_buffer_count = sum( - 1 - for key in self.buffers - if key[0] == execution_context.instance_uuid - and key[1] == execution_context.workspace_uuid - and key[2] == execution_context.placement_generation + scope_key = ( + execution_context.instance_uuid, + execution_context.workspace_uuid, + execution_context.placement_generation, + ) + workspace_buffer_count = self._buffer_counts_by_scope.get( + scope_key, + 0, ) if len(self.buffers) >= self.max_buffers or workspace_buffer_count >= self.max_buffers_per_workspace: bypass_aggregation = True @@ -210,6 +216,7 @@ class MessageAggregator: messages=[pending_msg], ) self.buffers[aggregation_key] = buffer + self._buffer_counts_by_scope[scope_key] = workspace_buffer_count + 1 else: if buffer.execution_context != execution_context: raise ExecutionContextMismatchError('Aggregation buffer ExecutionContext changed for the same key') @@ -282,6 +289,12 @@ class MessageAggregator: if buffer.execution_context != execution_context: raise ExecutionContextMismatchError('Timer ExecutionContext does not match the aggregation buffer') self.buffers.pop(aggregation_key) + scope_key = aggregation_key[:3] + scope_count = self._buffer_counts_by_scope.get(scope_key, 0) + if scope_count <= 1: + self._buffer_counts_by_scope.pop(scope_key, None) + else: + self._buffer_counts_by_scope[scope_key] = scope_count - 1 if not buffer.messages: return diff --git a/src/langbot/pkg/provider/tools/loaders/mcp.py b/src/langbot/pkg/provider/tools/loaders/mcp.py index 08fefb4aa..2084bd154 100644 --- a/src/langbot/pkg/provider/tools/loaders/mcp.py +++ b/src/langbot/pkg/provider/tools/loaders/mcp.py @@ -233,8 +233,6 @@ class MCPToolCallTimeoutError(TimeoutError): class RuntimeMCPSession: """运行时 MCP 会话""" - _FENCE_POLL_INTERVAL = 5.0 - ap: app.Application server_name: str @@ -362,15 +360,6 @@ class RuntimeMCPSession: if binding.instance_uuid != self.execution_context.instance_uuid: raise WorkspaceInvariantError('MCP session instance does not match the active Workspace binding') - async def _monitor_execution_fence(self) -> None: - """Poll the placement fence while an MCP transport is idle.""" - - while not self._shutdown_event.is_set(): - await asyncio.sleep(self._FENCE_POLL_INTERVAL) - if self._shutdown_event.is_set(): - return - await self._assert_execution_active() - async def _sleep_with_execution_fence(self, delay: float) -> None: """Back off without reconnecting after the captured placement expires.""" @@ -576,16 +565,13 @@ class RuntimeMCPSession: monitor_task = asyncio.create_task(self._box_stdio_runtime.monitor_process_health()) shutdown_task = asyncio.create_task(self._shutdown_event.wait()) reconnect_task = asyncio.create_task(self._reconnect_event.wait()) - fence_task = asyncio.create_task(self._monitor_execution_fence()) done, pending = await asyncio.wait( - [shutdown_task, monitor_task, reconnect_task, fence_task], + [shutdown_task, monitor_task, reconnect_task], return_when=asyncio.FIRST_COMPLETED, ) for task in pending: task.cancel() await asyncio.gather(*pending, return_exceptions=True) - if fence_task in done and not self._shutdown_event.is_set(): - fence_task.result() if reconnect_task in done and not self._shutdown_event.is_set(): self._reconnect_event.clear() self.ap.logger.info( @@ -621,16 +607,13 @@ class RuntimeMCPSession: else: shutdown_task = asyncio.create_task(self._shutdown_event.wait()) reconnect_task = asyncio.create_task(self._reconnect_event.wait()) - fence_task = asyncio.create_task(self._monitor_execution_fence()) done, pending = await asyncio.wait( - [shutdown_task, reconnect_task, fence_task], + [shutdown_task, reconnect_task], return_when=asyncio.FIRST_COMPLETED, ) for task in pending: task.cancel() await asyncio.gather(*pending, return_exceptions=True) - if fence_task in done and not self._shutdown_event.is_set(): - fence_task.result() if reconnect_task in done and not self._shutdown_event.is_set(): self._reconnect_event.clear() self.ap.logger.info( @@ -1550,6 +1533,8 @@ class MCPLoader(loader.ToolLoader): set[asyncio.Task], ] = {} self._host_dispatch_tasks: set[asyncio.Task] = set() + self._pending_projection_retirements: set[tuple[str, str, int]] = set() + self._projection_reconcile_task: asyncio.Task[None] | None = None config = getattr(getattr(ap, 'instance_config', None), 'data', {}) mcp_config = config.get('mcp', {}) if isinstance(config, dict) else {} raw_lifecycle_concurrency = mcp_config.get('lifecycle_concurrency', 16) if isinstance(mcp_config, dict) else 16 @@ -1691,7 +1676,59 @@ class MCPLoader(loader.ToolLoader): keys = tuple(self._session_keys_by_scope.pop(scope_key, ())) sessions = [session for key in keys if (session := self._sessions.pop(key, None)) is not None] await self._shutdown_sessions(sessions) - self._scope_generations.pop(scope_key[:2], None) + if self._scope_generations.get(scope_key[:2]) == scope_key[2]: + self._scope_generations.pop(scope_key[:2], None) + + def reconcile_execution_projection( + self, + instance_uuid: str, + active_generations: typing.Mapping[str, int], + *, + affected_workspace_uuids: typing.Iterable[str] | None = None, + ) -> None: + """Queue stale MCP scopes for one coalesced, bounded cleanup worker.""" + + affected = None if affected_workspace_uuids is None else set(affected_workspace_uuids) + for workspace_scope, generation in tuple(self._scope_generations.items()): + scoped_instance_uuid, workspace_uuid = workspace_scope + if scoped_instance_uuid != instance_uuid: + continue + if affected is not None and workspace_uuid not in affected: + continue + if active_generations.get(workspace_uuid) == generation: + continue + self._pending_projection_retirements.add((*workspace_scope, generation)) + + if not self._pending_projection_retirements: + return + if self._projection_reconcile_task is not None and not self._projection_reconcile_task.done(): + return + task = asyncio.create_task( + self._drain_projection_retirements(), + name='mcp-projection-reconcile', + ) + self._projection_reconcile_task = task + task.add_done_callback(self._projection_reconcile_done) + + async def _drain_projection_retirements(self) -> None: + while self._pending_projection_retirements: + scope_key = next(iter(self._pending_projection_retirements)) + self._pending_projection_retirements.discard(scope_key) + await self._retire_runtime_scope(scope_key) + + def _projection_reconcile_done( + self, + completed: asyncio.Task[None], + ) -> None: + if self._projection_reconcile_task is completed: + self._projection_reconcile_task = None + if completed.cancelled(): + return + exception = completed.exception() + if exception is not None: + self.ap.logger.error( + f'MCP projection reconciliation failed: {exception}', + ) async def _observe_execution_context( self, @@ -1713,6 +1750,13 @@ class MCPLoader(loader.ToolLoader): async def _reset_runtime_state(self) -> None: """Cancel host tasks and close sessions before reload or shutdown.""" + projection_task = self._projection_reconcile_task + self._projection_reconcile_task = None + self._pending_projection_retirements.clear() + if projection_task is not None and not projection_task.done(): + projection_task.cancel() + await asyncio.gather(projection_task, return_exceptions=True) + dispatch_tasks = tuple(self._host_dispatch_tasks) self._host_dispatch_tasks.clear() for task in dispatch_tasks: diff --git a/tests/unit_tests/cloud/test_directory_projection.py b/tests/unit_tests/cloud/test_directory_projection.py index 38af7ce2f..f19e9fe0f 100644 --- a/tests/unit_tests/cloud/test_directory_projection.py +++ b/tests/unit_tests/cloud/test_directory_projection.py @@ -3,6 +3,7 @@ from __future__ import annotations import datetime import logging from types import SimpleNamespace +from unittest.mock import Mock import pytest import sqlalchemy @@ -181,6 +182,12 @@ def _delta( async def test_initial_snapshot_projects_core_owned_rows(projection_context): application, session_factory = projection_context + reconcile_execution_projection = Mock() + application.tool_mgr = SimpleNamespace( + mcp_tool_loader=SimpleNamespace( + reconcile_execution_projection=reconcile_execution_projection, + ) + ) service = DirectoryProjectionService( application, _Provider([_snapshot(1)]), @@ -212,6 +219,11 @@ async def test_initial_snapshot_projects_core_owned_rows(projection_context): assert state is not None assert state.cursor == 1 assert state.snapshot_coverage_cursor == 1 + reconcile_execution_projection.assert_called_once_with( + INSTANCE_UUID, + {WORKSPACE_UUID: 1}, + affected_workspace_uuids=None, + ) async def test_same_cursor_equivocation_and_rollback_fail_closed(projection_context): @@ -297,6 +309,12 @@ async def test_archived_and_absent_workspaces_are_execution_fenced(projection_co async def test_event_poll_fetches_workspace_delta_and_records_receipt(projection_context): application, session_factory = projection_context + reconcile_execution_projection = Mock() + application.tool_mgr = SimpleNamespace( + mcp_tool_loader=SimpleNamespace( + reconcile_execution_projection=reconcile_execution_projection, + ) + ) event = DirectoryEvent( cursor=2, uuid='40000000-0000-0000-0000-000000000001', @@ -320,6 +338,7 @@ async def test_event_poll_fetches_workspace_delta_and_records_receipt(projection ) service = DirectoryProjectionService(application, provider, INSTANCE_UUID) await service.initialize() + reconcile_execution_projection.reset_mock() await service.sync_once() @@ -335,6 +354,11 @@ async def test_event_poll_fetches_workspace_delta_and_records_receipt(projection assert inbox.applied_at is not None assert provider.snapshot_calls == 1 assert provider.delta_calls == 1 + reconcile_execution_projection.assert_called_once_with( + INSTANCE_UUID, + {WORKSPACE_UUID: 1}, + affected_workspace_uuids={WORKSPACE_UUID}, + ) async def test_directory_delta_does_not_skip_unfetched_event_cursors(projection_context): diff --git a/tests/unit_tests/pipeline/test_aggregator.py b/tests/unit_tests/pipeline/test_aggregator.py index e8bdb3709..4b8e5f059 100644 --- a/tests/unit_tests/pipeline/test_aggregator.py +++ b/tests/unit_tests/pipeline/test_aggregator.py @@ -875,6 +875,57 @@ class TestMessageAggregatorWorkspaceIsolation: assert {key[1] for key in agg.buffers} == {'workspace-a', 'workspace-b'} await agg.flush_all() + @pytest.mark.asyncio + async def test_new_buffer_uses_scope_counter_without_global_scan(self): + class NoGlobalIterationDict(dict): + def __iter__(self): + raise AssertionError('aggregation admission scanned all buffers') + + def items(self): + raise AssertionError('aggregation admission scanned all buffers') + + def values(self): + raise AssertionError('aggregation admission scanned all buffers') + + app = make_aggregator_app() + enable_aggregation(app) + agg = get_aggregator_module().MessageAggregator(app) + agg.max_buffers = 2_000 + agg.max_buffers_per_workspace = 2_000 + existing = { + ( + 'instance-test', + f'workspace-{index}', + 1, + 'bot', + 'pipeline', + 'person', + index, + ): object() + for index in range(1_000) + } + agg.buffers = NoGlobalIterationDict(existing) + agg._buffer_counts_by_scope = {key[:3]: 1 for key in existing} + context = execution_context( + 'workspace-target', + pipeline_uuid='test-pipeline', + ) + + await agg.add_message(**scoped_message_kwargs(context)) + + key = aggregation_key( + context, + pipeline_uuid='test-pipeline', + ) + assert key in agg.buffers + assert agg._buffer_counts_by_scope[key[:3]] == 1 + timer_task = agg.buffers[key].timer_task + assert timer_task is not None + timer_task.cancel() + await asyncio.gather(timer_task, return_exceptions=True) + await agg._flush_buffer(key, context) + assert key[:3] not in agg._buffer_counts_by_scope + @pytest.mark.asyncio async def test_same_launcher_in_two_bots_uses_separate_buffers(self): app = make_aggregator_app() diff --git a/tests/unit_tests/provider/test_mcp_resources.py b/tests/unit_tests/provider/test_mcp_resources.py index efc039cce..62d179a55 100644 --- a/tests/unit_tests/provider/test_mcp_resources.py +++ b/tests/unit_tests/provider/test_mcp_resources.py @@ -558,27 +558,72 @@ async def test_mcp_resource_cache_is_not_served_to_stale_generation(): @pytest.mark.asyncio -async def test_mcp_idle_lifecycle_stops_without_retry_after_generation_bump(): - session = _connected_session() - session.server_config.update({'mode': 'remote', 'url': 'https://example.com/mcp'}) - session._FENCE_POLL_INTERVAL = 0 - session._init_remote_server = AsyncMock() - session.refresh = AsyncMock() - session._assert_execution_active = AsyncMock( - side_effect=[ - None, - None, - None, - WorkspaceGenerationMismatchError('generation changed while idle'), - ] +async def test_directory_projection_retires_idle_mcp_scope_without_db_poll(): + loader = MCPLoader(_app()) + sessions = [] + for index in range(100): + context = ExecutionContext( + instance_uuid='instance-a', + workspace_uuid=f'workspace-{index}', + placement_generation=1, + ) + session = RuntimeMCPSession( + f'server-{index}', + {'uuid': f'srv-{index}', 'mode': 'remote'}, + True, + loader.ap, + context, + ) + session.shutdown = AsyncMock() + loader._register_session(context, session.server_name, session) + sessions.append(session) + + loader.reconcile_execution_projection('instance-a', {}) + reconcile_task = loader._projection_reconcile_task + assert reconcile_task is not None + assert len(loader._pending_projection_retirements) == 100 + + # A second projection coalesces into the same worker instead of creating + # one timer or task per Workspace. + loader.reconcile_execution_projection('instance-a', {}) + assert loader._projection_reconcile_task is reconcile_task + + await asyncio.wait_for(reconcile_task, timeout=1) + + assert loader.sessions == {} + assert loader._scope_generations == {} + assert loader._pending_projection_retirements == set() + assert sum(session.shutdown.await_count for session in sessions) == 100 + loader.ap.workspace_service.get_execution_binding.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_directory_projection_keeps_matching_and_unaffected_mcp_scopes(): + loader = MCPLoader(_app()) + matching = _connected_session() + other_context = ExecutionContext( + instance_uuid='instance-a', + workspace_uuid='workspace-b', + placement_generation=1, ) + unaffected = _connected_session( + name='other', + uuid='srv-2', + execution_context=other_context, + ) + _register_session(loader, matching) + _register_session(loader, unaffected) - await session._lifecycle_loop_with_retry() + loader.reconcile_execution_projection( + 'instance-a', + {'workspace-a': 1}, + affected_workspace_uuids={'workspace-a'}, + ) + await asyncio.sleep(0) - session._init_remote_server.assert_awaited_once_with() - assert session.status == MCPSessionStatus.ERROR - assert session.error_message == 'Workspace execution binding is stale' - assert session._shutdown_event.is_set() + assert loader.get_session(TEST_EXECUTION_CONTEXT, 'docs') is matching + assert loader.get_session(other_context, 'other') is unaffected + assert loader._projection_reconcile_task is None @pytest.mark.asyncio diff --git a/tests/unit_tests/utils/test_cloud_runtime_soak.py b/tests/unit_tests/utils/test_cloud_runtime_soak.py index c0367cc54..a3c5cbd51 100644 --- a/tests/unit_tests/utils/test_cloud_runtime_soak.py +++ b/tests/unit_tests/utils/test_cloud_runtime_soak.py @@ -397,6 +397,7 @@ def test_evaluate_gate_detects_restart_circuit_and_stuck_launch() -> None: 0, **{ f'{prefix}.active_launches': 1, + f'{prefix}.gate_waiters': 2, f'{prefix}.half_open_probe_inflight': 1, f'{prefix}.open_remaining_seconds': 60, f'{prefix}.circuit_open_total': 0, @@ -406,6 +407,7 @@ def test_evaluate_gate_detects_restart_circuit_and_stuck_launch() -> None: 60, **{ f'{prefix}.active_launches': 1, + f'{prefix}.gate_waiters': 2, f'{prefix}.half_open_probe_inflight': 1, f'{prefix}.open_remaining_seconds': 1, f'{prefix}.circuit_open_total': 1, @@ -422,10 +424,43 @@ def test_evaluate_gate_detects_restart_circuit_and_stuck_launch() -> None: assert any('circuit_open_total by 1' in failure for failure in result.failures) assert any('active_launches above zero' in failure for failure in result.failures) + assert any('gate_waiters above zero' in failure for failure in result.failures) assert any('half_open_probe_inflight above zero' in failure for failure in result.failures) assert any('open_remaining_seconds above zero' in failure for failure in result.failures) +def test_evaluate_gate_detects_stuck_mcp_projection_cleanup() -> None: + prefix = 'body.resources.runtimes' + state = _state( + 'endpoint', + [ + _sample( + 0, + **{ + f'{prefix}.mcp_projection_retirements': 3, + f'{prefix}.mcp_projection_reconcile_active': 1, + }, + ), + _sample( + 60, + **{ + f'{prefix}.mcp_projection_retirements': 1, + f'{prefix}.mcp_projection_reconcile_active': 1, + }, + ), + ], + ) + + result = soak.evaluate_gate( + [state], + analysis_start_seconds=0, + thresholds=_thresholds(), + ) + + assert any('mcp_projection_retirements above zero' in failure for failure in result.failures) + assert any('mcp_projection_reconcile_active above zero' in failure for failure in result.failures) + + def test_evaluate_gate_detects_event_loop_stall_and_sustained_lag() -> None: prefix = 'body.resources.event_loop' samples = [ diff --git a/uv.lock b/uv.lock index a00c9154c..074887671 100644 --- a/uv.lock +++ b/uv.lock @@ -2116,7 +2116,7 @@ requires-dist = [ { name = "ebooklib", specifier = ">=0.18" }, { name = "gewechat-client", specifier = ">=0.1.5" }, { name = "html2text", specifier = ">=2024.2.26" }, - { name = "langbot-plugin", git = "https://github.com/langbot-app/langbot-plugin-sdk.git?rev=a5a96b302a5808af84bbdd28833ce050176f87e9" }, + { name = "langbot-plugin", git = "https://github.com/langbot-app/langbot-plugin-sdk.git?rev=c7893f8f94e83cb09ebcc98e25490fa699b0fb69" }, { name = "langchain", specifier = ">=1.3.9" }, { name = "langchain-core", specifier = ">=1.3.3" }, { name = "langchain-text-splitters", specifier = ">=1.1.2" }, @@ -2183,7 +2183,7 @@ dev = [ [[package]] name = "langbot-plugin" version = "0.4.18" -source = { git = "https://github.com/langbot-app/langbot-plugin-sdk.git?rev=a5a96b302a5808af84bbdd28833ce050176f87e9#a5a96b302a5808af84bbdd28833ce050176f87e9" } +source = { git = "https://github.com/langbot-app/langbot-plugin-sdk.git?rev=c7893f8f94e83cb09ebcc98e25490fa699b0fb69#c7893f8f94e83cb09ebcc98e25490fa699b0fb69" } dependencies = [ { name = "aiofiles" }, { name = "aiohttp" },