From a54f75e5e33775d5dc91f3058f5505a56b498bf2 Mon Sep 17 00:00:00 2001 From: jesse-ux Date: Thu, 15 Oct 2026 13:33:06 +0800 Subject: [PATCH] fix(capacity): limit each user and queue heavy compute briefly. Consult, rectification turns, and report generation stop before a credit reservation when the product caps are hit. A full heavy-compute gate waits up to 8 seconds, and one consultation retries that 429 inside the same request. Co-Authored-By: Grok (agent loop) --- CHANGELOG.md | 6 + deploy/README.md | 2 +- docs/BUG_HISTORY.md | 31 ++ docs/operations/beta-invite-codes.md | 21 ++ docs/tasks/PROGRESS-beta-capacity-20261014.md | 43 +++ docs/testing/beta-capacity-20261014.md | 6 + frontend/DESIGN.md | 2 +- frontend/docs/VOICE.md | 4 + frontend/scripts/research/beta-load-smoke.mts | 277 ++++++++++++++++++ frontend/src/app/api/consult/route.ts | 32 +- .../src/app/api/rectification/agent/route.ts | 19 ++ frontend/src/app/api/reports/route.ts | 12 +- frontend/src/components/chat-message-row.tsx | 1 + .../src/lib/consultation-activity-labels.ts | 2 + frontend/src/lib/consultation-agent-events.ts | 3 + frontend/src/lib/consultation-run-timeline.ts | 2 +- .../src/lib/consultation-workflow-retry.ts | 57 ++++ .../src/lib/personal-report-route-core.ts | 24 +- frontend/src/lib/request-rate-limit.ts | 75 ++++- frontend/src/lib/stream-agent-response.ts | 3 +- frontend/src/mastra/consultation-tools.ts | 57 +++- frontend/src/mastra/consultation-workflow.ts | 12 +- .../tests/consultation-workflow-retry.test.ts | 102 +++++++ frontend/tests/personal-report-api.test.ts | 13 + frontend/tests/request-rate-limit.test.ts | 100 +++++++ scripts/api_heavy_compute_gate.py | 98 +++++-- tests/test_api_chart_subprocess_gate.py | 10 +- tests/test_api_heavy_compute_gate.py | 91 +++++- 28 files changed, 1035 insertions(+), 70 deletions(-) create mode 100644 docs/operations/beta-invite-codes.md create mode 100644 docs/tasks/PROGRESS-beta-capacity-20261014.md create mode 100644 docs/testing/beta-capacity-20261014.md create mode 100644 frontend/scripts/research/beta-load-smoke.mts create mode 100644 frontend/src/lib/consultation-workflow-retry.ts create mode 100644 frontend/tests/consultation-workflow-retry.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 9123dd63..d148c6db 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # 印度占星 Skill 更新日志 +## 2026-10-15 — 内测容量：问得太密会停一下，人多时先排一小会儿 + +- 同一个人短时间问得太密、校正答得太密，或一小时里报告生成太多，会先停一下，这次不扣点。 +- 同时算的人多过机器能接的两路时，先排一小会儿。排上了继续；排不上会说人有点多，并把这一轮预留的点数退回。 +- Skill 版本不变（BUG-1355、BUG-1356）。 + ## 2026-10-15 — 合盘更快（待合入 staging） - 合盘排两张盘时关掉用不上的外部概览，两张盘一起排。个人报告在没装参照引擎时不再为注定失败的回放反复起进程。轻重任务要不要分开限流，这一单只给了模拟表，还没改规则。 diff --git a/deploy/README.md b/deploy/README.md index ffc9ab62..a8926333 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -37,7 +37,7 @@ The API image `WORKDIR` is `/app` (`deploy/railway-api.Dockerfile`). Runtime fil The volume is not a backup domain. Chart cache is a TTL'd compute cache (default 900s) and is cheaper to rebuild than to archive. Async / high-rigor job state is short-lived (default TTL 3600s) and is not a restore unit. Restart persistence is the only reason the volume exists. Staging encrypted backups remain PostgreSQL-only via `deploy/backup-staging-postgres.sh`; do not add `api_scratch` to that helper. Project-prefixed volume names are `jyotisha-staging_api_scratch` and `jyotisha-production_api_scratch`. -Heavy rectification scans and high-rigor workflows share a process-wide fail-fast concurrency gate (`JYOTISH_HEAVY_COMPUTE_CONCURRENCY`, default `2` to match the 2 vCPU host). Saturated requests return HTTP 429 with `Retry-After` and are not queued. Health checks and other light routes do not take a slot. +Heavy rectification scans and high-rigor workflows share a process-wide concurrency gate (`JYOTISH_HEAVY_COMPUTE_CONCURRENCY`, default `2` to match the 2 vCPU host). A saturated request waits up to 8 seconds if fewer than 4 requests are already waiting, then returns HTTP 429 with `Retry-After`. Health checks and other light routes do not take a slot. ## DNS and Supabase Auth diff --git a/docs/BUG_HISTORY.md b/docs/BUG_HISTORY.md index 455b974d..1b715132 100644 --- a/docs/BUG_HISTORY.md +++ b/docs/BUG_HISTORY.md @@ -36257,3 +36257,34 @@ - 复发自：无。 - 修复版本：未修。分支 `codex/answer-effort-ab-20261013` 只记录。原先写在 BUG-1351，该号已被数据层占用；按本文件规则保留较早的 BUG-1351，本条改为 BUG-1354，接在 BUG-1353 之后。 +## BUG-1355 | 对话、校正回合和报告生成没有按用户限流 + +- 状态：mitigated +- 首次发现 / 最近更新：2026-10-14 / 2026-10-15 +- 来源：issue #76。任务书 `docs/tasks/TASK-beta-capacity-20261014.md`。 +- 现象：一个人可以连续打对话、校正回合或报告生成。报告只有每天 5 份的上限，没有按小时的上限。对话和校正没有按这个人计数。 +- 触发条件：已登录后调用 `POST /api/consult`、`POST /api/rectification/agent` 或 `POST /api/reports`。开案 `POST /api/rectification/cases/open` 不计数。 +- 根因：`requestRateLimits` 只覆盖地点搜索、时区、今日星语、星历、合盘和会话写入。重计算闸是全进程共享的，挡不住同一个人把模型调用打满。 +- 修复：对话 10 次/分钟且 120 次/小时，校正回合 5 次/分钟且 40 次/小时，报告生成另加 3 次/小时（每天 5 份仍在）。两档都还有余量才放行；拒绝时不写入任何一档，也不预留点数。响应是 429、`code: rate_limited` 和 `Retry-After`。 +- 验证：`frontend/tests/request-rate-limit.test.ts`；报告核心测试「hourly rate limit returns 429 and does not reserve」。 +- 防复发：上述测试锁住桶的边界、拒绝时不记入另一档，以及三条路由在预留点数之前判断。开案路由源码不得调用这组限流。 +- 相关记录：BUG-723（校正客户端已有的退避，本单不改）、BUG-1302（预热仍不占重计算名额）。 +- 复发自：无 +- 修复版本：分支 `codex/beta-capacity-20261014`。 + +## BUG-1356 | 重计算满了立刻拒绝，对话也不再试 + +- 状态：mitigated +- 首次发现 / 最近更新：2026-10-14 / 2026-10-15 +- 来源：issue #76。任务书 `docs/tasks/TASK-beta-capacity-20261014.md`。 +- 现象：同时只有 2 个重计算名额。第 3 个请求马上收到 429。普通对话把这次 429 当成这一轮失败，不按 `Retry-After` 再试。 +- 触发条件：已有 2 个请求占着 `/api/consultation_workflow` 或校正打分，又来一个对话取证。 +- 根因：闸门用非阻塞信号量，满了就抛 `HeavyComputeBusy`。对话侧把 HTTP 429 记成 `workflow_rate_limited` 后直接失败。 +- 修复：名额满时最多再等 8 秒，同时等待的人最多 4 个。等不到或等待的人已经满，仍抛 `HeavyComputeBusy`，`Retry-After` 仍是原来的秒数。对话对这种 429 最多再试 2 次，几次等待加起来不超过 15 秒。等待时进度行写「人有点多，排一下队」。仍失败则沿用原来的取消预留退点，只走一次，并告诉用户这一轮没排上、点数已退回。并发上限仍是 2。 +- 验证：`tests/test_api_heavy_compute_gate.py` 的排队拿到名额、等待者满立刻拒绝、超时仍 429；`frontend/tests/consultation-workflow-retry.test.ts`。 +- 防复发：上述测试。健康检查仍不进闸。校正客户端原有的退避（BUG-723）没有改。 +- 相关记录：BUG-717、BUG-723、BUG-1302、BUG-1345、BUG-1355。 +- 复发自：无 +- 修复版本：分支 `codex/beta-capacity-20261014`。 + + diff --git a/docs/operations/beta-invite-codes.md b/docs/operations/beta-invite-codes.md new file mode 100644 index 00000000..29a74ef5 --- /dev/null +++ b/docs/operations/beta-invite-codes.md @@ -0,0 +1,21 @@ +# 内测邀请码（用兑换码） + +后台路径：`/admin/codes`（兑换码）。需要管理权限。不要把真实兑换码写进仓库、进度或聊天记录。完整码只在生成成功的那一次弹窗里出现，关掉就不能再看，列表里只剩掩码。 + +## 批量生成 + +1. 打开兑换码页，点「批量生成」。 +2. 「每个点数」是这一批每个码到账的点数。表单初始值是 10，可以改。内测每码给多少点由当次生成的人填写，产品以后可以改这个数。 +3. 「数量」一次 1 到 100 个。同一批的点数、到期时间、备注相同。 +4. 到期时间可以空着。备注用来区分这一批，例如内测第几批，不要写别人的个人资料。 +5. 生成后立刻把完整码交给对应的人。页面会提示关闭后无法再看。 + +## 查谁用了 + +列表有掩码、点数、状态、到期、备注、兑换账户、兑换时间、撤销时间。状态可以筛：可用、已过期、已兑换、已撤销。兑换账户和兑换时间就是谁用过。 + +## 作废 + +还没兑换、也没撤销的码，行上有「撤销」。撤销后不能再兑。已经兑换的不能从这里追回已经到账的点数。 + +支付在内测期间保持关闭。这页已经能批量生成、查看使用者和作废，本单不另加功能。 diff --git a/docs/tasks/PROGRESS-beta-capacity-20261014.md b/docs/tasks/PROGRESS-beta-capacity-20261014.md new file mode 100644 index 00000000..39a02628 --- /dev/null +++ b/docs/tasks/PROGRESS-beta-capacity-20261014.md @@ -0,0 +1,43 @@ +# 进度 · 内测容量（2026-10-15） + +- 分支：`codex/beta-capacity-20261014` +- 基线：开工时 `origin/staging` `ef49257b734678ff944a840076347baa02d0323c` +- 任务书：`docs/tasks/TASK-beta-capacity-20261014.md` +- 前置：`origin/codex/node-runtime-resources-20261013` 与 `origin/codex/rect-scoring-perf-20261013` 都是该 staging 的祖先（`merge-base --is-ancestor` 退出码 0）。限流文件里已有过期清理。 + +## 做了什么 + +- T1：对话 10 次/分钟 + 120 次/小时，校正回合 5 次/分钟 + 40 次/小时，报告 3 次/小时（每天 5 份不变）。拒绝发生在预留点数之前，429 + `rate_limited` + `Retry-After`。开案路由不计数。 +- T2：重计算并发仍是 2。满了最多等 8 秒，同时等待最多 4 个；等不到或等待已满仍是 429，`Retry-After` 不变。对话对这种 429 最多再试 2 次，等待合计不超过 15 秒。进度行「人有点多，排一下队」，失败句「人有点多，这一轮没排上，点数已经退回。」退点仍走原来的取消预留，代理路径用已有的一次结算。 +- T3：后台已能批量生成（1–100 个、每码点数自填、初始 10）、按兑换账户查看、撤销未使用的码。说明在 `docs/operations/beta-invite-codes.md`。没有扩功能。 +- T4：脚本 `frontend/scripts/research/beta-load-smoke.mts`。不从 `frontend/tests` 导入。本机没有跑。 +- T5：BUG-1355、BUG-1356，CHANGELOG、VOICE、DESIGN、真机两条。 + +## 断言 + +旧测试仍要求「名额被占着就拒绝」，没有改拒绝码和 `Retry-After` 的期望值。 + +| 位置 | 原值 | 新值 | 原因 | +| --- | --- | --- | --- | +| `tests/test_api_heavy_compute_gate.py` 立刻拒绝的夹具和单测 | `reset()` / `reset(limit=1, retry_after_seconds=3)` | 增加 `wait_seconds=0` | 默认改为最多等 8 秒。这些测试继续锁 429 和 `Retry-After`，排队另有新测试 | +| `tests/test_api_chart_subprocess_gate.py` 两处占着名额再请求 | `reset(limit=1)` | `reset(limit=1, wait_seconds=0)` | 同上。仍锁同步排盘被拒绝、日志不出现请求正文 | + +## 偏离与环境缺口 + +- T0：仓库文档里的目标规格，staging 与生产都是 2 vCPU / 4 GB，代码默认并发 2。这台 Windows 没有 staging 的登录壳，没有采到机器上的实际核数、内存和 `JYOTISH_HEAVY_COMPUTE_CONCURRENCY`。不能把文档规格当成已经登主机核对过。 +- T4：没有 `STAGING_TEST_OTP`，没有对 staging 或生产发压测。改前、改后的并发数字都没有。服务器 CPU / 内存同样没有采样。脚本留给验收方在有测试验证码的环境重跑。20 并发档没有跑。 +- 旧版对话（不走代理运行时）会重试，但进度行要等回答流开始才有，所以那条路径等待时看不到「排一下队」。现在的对话走代理运行时，进度行会换字。任务书允许把排队文案收成只多等。 +- 没有改数据库，没跑 `test:db`。没有跑前端全量 `npm test`，没有跑 Python 快速门。 +- `deploy/README.md` 有一句仍写「满了不排队」。改成了最多等 8 秒、等待不超过 4 个，避免操作说明和代码相反。没有改部署配置。 + +## 自测 + +| 检查 | 命令 | 结果数字 | +| --- | --- | --- | +| Python 闸门 | `.venv\Scripts\python.exe -m pytest tests/test_api_heavy_compute_gate.py tests/test_api_chart_subprocess_gate.py -q` | 17 通过（8 + 9），0 失败 | +| 前端相关测试 | `node node_modules/tsx/dist/cli.mjs --test tests/request-rate-limit.test.ts tests/consultation-workflow-retry.test.ts tests/personal-report-api.test.ts tests/runtime-sources-no-test-imports.test.ts` | 76 通过，0 失败 | +| 类型 | `frontend\node_modules\.bin\tsc --noEmit -p .` | 退出码 0 | +| 前端 lint | `npm run lint` | 0 错误，123 条原有警告 | +| 压测 | 未执行 `frontend/scripts/research/beta-load-smoke.mts` | 本机没有 `STAGING_TEST_OTP`。没有改前/改后并发数字，没有服务器 CPU/内存 | + +没有改数据库，没跑 `test:db`。没有跑前端全量 `npm test`，没有跑 Python 快速门。 diff --git a/docs/testing/beta-capacity-20261014.md b/docs/testing/beta-capacity-20261014.md new file mode 100644 index 00000000..862a6016 --- /dev/null +++ b/docs/testing/beta-capacity-20261014.md @@ -0,0 +1,6 @@ +# 内测容量 · 真机两条（2026-10-14） + +用测试账号。不要把出生资料、姓名、邮箱写进记录。 + +1. 快速连发触发限流。登录后在一分钟里连续发送超过 10 次普通对话。预期：出现「这一会儿问得有点密，先停一下。」这一轮不扣点。等提示里的时间过去再发一次，可以正常问。 +2. 排队提示出现与消失。需要两个人同时占着计算（或用压测脚本打满两个名额）时，第三个人发普通对话。预期：进度行先出现「人有点多，排一下队」，排上之后这句消失，回到正在计算。若一直排不上，出现「人有点多，这一轮没排上，点数已经退回。」余额与发送前一致。 diff --git a/frontend/DESIGN.md b/frontend/DESIGN.md index 74d2c6f0..006fbc0b 100644 --- a/frontend/DESIGN.md +++ b/frontend/DESIGN.md @@ -907,7 +907,7 @@ or user IDs. Agent 的 live 标记只有 `InlineSpinner` 一种。曾经并存的 canvas 小球（`thinking-orbs`）已移除，不得再引入第二种 live 标记。 -**普通对话发出即有反馈（2026-09-28，TASK-consult-first-frame-and-pacing，BUG-1074）。** 按下发送的那一帧，回答行就带着步骤栏，唯一一行是排队行「收到，正在看你的问题…」（`QUEUED_TIMELINE_ROW`，行内 `InlineSpinner` + shimmer，summary 写「正在分析」）。服务端流的第一个字节也是这句（`activity` 事件，phase `received`），所以那一帧到达时行不换字。不再有「分类结果回来之前整行静默」：只有服务端判定为寒暄的回复（`run.completed` 带 `responseKind: "smalltalk"`）才把步骤栏收掉、只留一句话，寒暄前会闪约一秒这一行，属于流式生成中的正常形态。首事件之后的行照旧由服务端事件推动；不新增第二种 live 标记，不加 spinner 面板或骨架。 +**普通对话发出即有反馈（2026-09-28，TASK-consult-first-frame-and-pacing，BUG-1074）。** 按下发送的那一帧，回答行就带着步骤栏，唯一一行是排队行「收到，正在看你的问题…」（`QUEUED_TIMELINE_ROW`，行内 `InlineSpinner` + shimmer，summary 写「正在分析」）。服务端流的第一个字节也是这句（`activity` 事件，phase `received`），所以那一帧到达时行不换字。不再有「分类结果回来之前整行静默」：只有服务端判定为寒暄的回复（`run.completed` 带 `responseKind: "smalltalk"`）才把步骤栏收掉、只留一句话，寒暄前会闪约一秒这一行，属于流式生成中的正常形态。首事件之后的行照旧由服务端事件推动；不新增第二种 live 标记，不加 spinner 面板或骨架。重计算名额满了时，计算那一行先换成「人有点多，排一下队」（`activity` phase `compute-queue`，仍是原来的 live 行），排上之后回到「正在计算本命盘…」。排不上则整轮失败，文案是「人有点多，这一轮没排上，点数已经退回。」 校正面的所有等待复用行内等待：进入前的 hydration 在揭幕之前完成，进入后唯一的等待形态是时间线 live 行（含「正在准备下一个问题…」这一条独立 live 行）。打字回答的 live 行从发出那一帧起就是阶段句「收到，正在对照你的档案…」，按服务端 `turn.progress` 换成「正在记下这件事…」「正在重新对照盘面…」「正在准备下一个问题…」，属于流式生成中（BUG-1047）。区间交付卡只挂在最新那条采用旁白下面，不得留在更早的采集/区分题下。有未答的采集或选择题时卡仍在，「更像这个」置灰并写「先答完上面这道，再选时间」，不得整卡消失。卡上至多三列并排，相同性格句只写一次，点「更像这个」即采用该列分钟，按钮显示「正在采用…」或「已采用」。采用过程中整张卡留在原处，不得因 `busy` 卸掉。采用后若还有前事核对，服务端写一条自己的采用消息（「已按 HH:MM 排盘…」），核对题的四个选项挂在这条消息下，进行中那一行由它替换，不挂到上一题的回复下（BUG-1146）。采用后前事核对结束走 `verified_idle`：一行收尾文案跟在卡片下面、与助手列对齐，没有 live 行、没有重载、没有采用状态条。卡片与右栏细则见 §11、§12。 diff --git a/frontend/docs/VOICE.md b/frontend/docs/VOICE.md index 8c16bc72..9d726ac1 100644 --- a/frontend/docs/VOICE.md +++ b/frontend/docs/VOICE.md @@ -480,3 +480,7 @@ Jyotisha 的可见文案是产品的一部分。正确性红线（真实性、 - 不写健康诊断、投资建议、法律意见；不写「一定」「必然」「注定」。 - 出生时间没确认时不用宫位，只按月亮所在星座写，caution 固定为「出生时间尚未确认，这里只看大方向，不依赖具体分钟。」 +## 人多与问得太密（2026-10-15） + +不写「系统繁忙」「请稍后重试」。计算要等空位时，进度行只写「人有点多，排一下队」。这一轮最终没排上：「人有点多，这一轮没排上，点数已经退回。」对话太密：「这一会儿问得有点密，先停一下。」校正回合太密：「这一会儿校正问得有点密，先停一下。」这一小时报告太多：「这一小时的报告已经够多了，先停一下。」 + diff --git a/frontend/scripts/research/beta-load-smoke.mts b/frontend/scripts/research/beta-load-smoke.mts new file mode 100644 index 00000000..2c65f38a --- /dev/null +++ b/frontend/scripts/research/beta-load-smoke.mts @@ -0,0 +1,277 @@ +/** + * Staging-only smoke for 10 or 20 concurrent consultations. + * + * Not imported by frontend/src. Reads STAGING_TEST_OTP from the environment + * and never prints it. Refuses a production host. Does not sample server + * CPU or memory; that needs a staging shell this process does not have. + * + * node node_modules/tsx/dist/cli.mjs scripts/research/beta-load-smoke.mts --users 10 + */ +const usersArg = process.argv.find((arg) => arg.startsWith("--users=")); +const usersFlag = process.argv.indexOf("--users"); +const requestedUsers = Number(usersArg ? usersArg.slice("--users=".length) : process.argv[usersFlag + 1]); +const baseUrl = (process.env.STAGING_BASE_URL || "https://staging.jyotisha.chat").replace(/\/$/, ""); +const otp = process.env.STAGING_TEST_OTP || ""; +const emailDomain = process.env.STAGING_TEST_EMAIL_DOMAIN || "example.test"; + +function refuse(message: string): never { + console.error(message); + process.exit(2); +} + +let host = ""; +try { + host = new URL(baseUrl).host; +} catch { + refuse("STAGING_BASE_URL is not a URL"); +} +if (host === "jyotisha.chat" || host === "www.jyotisha.chat" || host === "admin.jyotisha.chat") { + refuse("refusing production host"); +} +if (!/^\d{6}$/.test(otp)) refuse("STAGING_TEST_OTP is not set"); +if (requestedUsers !== 10 && requestedUsers !== 20) refuse("pass --users 10 or --users 20"); +if (!/^[a-z0-9.-]+$/.test(emailDomain)) refuse("STAGING_TEST_EMAIL_DOMAIN is not a plain domain"); + +type Sample = { + kind: "consult" | "rectification"; + status: number; + firstByteMs: number | null; + totalMs: number; + failed: boolean; + http429: boolean; + http5xx: boolean; +}; + +const samples: Sample[] = []; +const CONSULT_QUEUE_FAILED = "人有点多，这一轮没排上，点数已经退回。"; + +function percentile(values: number[], ratio: number): number | null { + if (values.length === 0) return null; + const sorted = [...values].sort((a, b) => a - b); + const index = Math.min(sorted.length - 1, Math.max(0, Math.ceil(sorted.length * ratio) - 1)); + return sorted[index] ?? null; +} + +async function request(jar: Map, path: string, init: RequestInit = {}): Promise { + const headers = new Headers(init.headers); + if (jar.size > 0) headers.set("cookie", [...jar].map(([key, value]) => `${key}=${value}`).join("; ")); + const response = await fetch(`${baseUrl}${path}`, { ...init, headers, redirect: "manual" }); + const setCookies = typeof response.headers.getSetCookie === "function" ? response.headers.getSetCookie() : []; + for (const line of setCookies) { + const pair = line.split(";")[0] ?? ""; + const eq = pair.indexOf("="); + if (eq > 0) jar.set(pair.slice(0, eq).trim(), pair.slice(eq + 1).trim()); + } + return response; +} + +async function timed(jar: Map, path: string, body: unknown, kind: Sample["kind"]): Promise { + const started = Date.now(); + let status = 0; + let firstByteMs: number | null = null; + let failed = false; + try { + const response = await request(jar, path, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }); + status = response.status; + const reader = response.body?.getReader(); + if (reader) { + let text = ""; + for (;;) { + const chunk = await reader.read(); + if (firstByteMs === null && (chunk.value || chunk.done)) firstByteMs = Date.now() - started; + if (chunk.value) text += new TextDecoder().decode(chunk.value); + if (chunk.done) break; + } + failed = text.includes('"type":"run.failed"') || text.includes(CONSULT_QUEUE_FAILED); + } else { + firstByteMs = Date.now() - started; + failed = status >= 400; + } + } catch { + status = 0; + failed = true; + } + samples.push({ + kind, + status, + firstByteMs, + totalMs: Date.now() - started, + failed, + http429: status === 429, + http5xx: status >= 500 || status === 0, + }); +} + +async function login(index: number): Promise | null> { + const jar = new Map(); + const email = `capacity-smoke-${index}@${emailDomain}`; + const sent = await request(jar, "/api/auth/email-otp/send-verification-otp", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ email, type: "sign-in" }), + }); + if (!sent.ok) return null; + const signed = await request(jar, "/api/auth/sign-in/email-otp", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ email, otp }), + }); + if (!signed.ok || jar.size === 0) return null; + return jar; +} + +async function prepareAccount(jar: Map): Promise { + const saved = await request(jar, "/api/account", { + method: "PATCH", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + name: "容量冒烟", + birth_date: "1990-06-15", + reported_birth_time: "12:00", + birth_time_source: "family_exact", + birth_time_period: null, + birth_time_clue: null, + uncertainty_before_minutes: 0, + uncertainty_after_minutes: 0, + country_code: "CN", + province_code: "110000", + city_code: "110000", + district_code: "110101", + latitude: 39.9042, + longitude: 116.4074, + timezone_offset: 8, + }), + }); + return saved.ok; +} + +async function oneUser(index: number): Promise { + const deleted: string[] = []; + const jar = await login(index); + if (!jar) { + samples.push({ kind: "consult", status: 0, firstByteMs: null, totalMs: 0, failed: true, http429: false, http5xx: true }); + return deleted; + } + if (!await prepareAccount(jar)) { + samples.push({ kind: "consult", status: 0, firstByteMs: null, totalMs: 0, failed: true, http429: false, http5xx: true }); + return deleted; + } + const models = await request(jar, "/api/models"); + const catalog = models.ok ? await models.json().catch(() => null) as { defaultModelId?: string } | null : null; + const modelId = catalog?.defaultModelId; + if (!modelId) { + samples.push({ kind: "consult", status: 0, firstByteMs: null, totalMs: 0, failed: true, http429: false, http5xx: true }); + return deleted; + } + const sessionId = crypto.randomUUID(); + const created = await request(jar, "/api/sessions", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + id: sessionId, + title: "容量冒烟", + theme: "general", + model_id: modelId, + messages: [], + session_type: "consultation", + rectification_case_id: null, + chart_profile_role: "self", + }), + }); + if (!created.ok) { + samples.push({ kind: "consult", status: created.status, firstByteMs: null, totalMs: 0, failed: true, http429: created.status === 429, http5xx: created.status >= 500 }); + return deleted; + } + const chart = { + year: 1990, + month: 6, + day: 15, + hour: 12, + minute: 0, + lat: 39.9042, + lon: 116.4074, + tz: 8, + city: "北京", + ayanamsa: "lahiri", + }; + await timed(jar, "/api/consult", { + ...chart, + requestId: crypto.randomUUID(), + sessionId, + modelId, + consultationMode: "verified_chart", + theme: "general", + question: "今年工作整体顺不顺？", + history: [], + }, "consult"); + await timed(jar, "/api/consult", { + ...chart, + requestId: crypto.randomUUID(), + sessionId, + modelId, + consultationMode: "verified_chart", + theme: "general", + question: "那接下来几个月呢？", + history: [], + }, "consult"); + if (index < 3) { + const opened = await request(jar, "/api/rectification/cases/open", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ intent: "homepage", requestId: crypto.randomUUID() }), + }); + const openedBody = opened.ok ? await opened.json().catch(() => null) as { caseId?: string; sessionId?: string } | null : null; + if (openedBody?.caseId && openedBody.sessionId) { + for (const message of ["我小时候搬家过一次。", "那次大概是上小学之前。"]) { + await timed(jar, "/api/rectification/agent", { + caseId: openedBody.caseId, + sessionId: openedBody.sessionId, + requestId: crypto.randomUUID(), + action: "message", + message, + }, "rectification"); + } + deleted.push(openedBody.sessionId); + } else { + samples.push({ + kind: "rectification", + status: opened.status, + firstByteMs: null, + totalMs: 0, + failed: true, + http429: opened.status === 429, + http5xx: opened.status >= 500 || opened.status === 0, + }); + } + } + const removed = await request(jar, `/api/sessions/${sessionId}`, { method: "DELETE" }); + if (removed.ok) deleted.push(sessionId); + return deleted; +} + +const startedAt = Date.now(); +const cleaned = (await Promise.all(Array.from({ length: requestedUsers }, (_, index) => oneUser(index)))).flat(); +const consults = samples.filter((sample) => sample.kind === "consult"); +const finalFailures = consults.filter((sample) => sample.failed || sample.http429 || sample.http5xx); +const firstBytes = consults.map((sample) => sample.firstByteMs).filter((value): value is number => value !== null); +const totals = consults.map((sample) => sample.totalMs); +const summary = { + users: requestedUsers, + host, + requests: samples.length, + consults: consults.length, + http429: samples.filter((sample) => sample.http429).length, + http5xx: samples.filter((sample) => sample.http5xx).length, + finalFailures: finalFailures.length, + finalFailureRatio: consults.length === 0 ? null : Number((finalFailures.length / consults.length).toFixed(4)), + firstByteP95Ms: percentile(firstBytes, 0.95), + totalP95Ms: percentile(totals, 0.95), + sessionsDeleted: cleaned.length, + serverSample: "not taken; this process has no staging docker stats", + elapsedMs: Date.now() - startedAt, +}; +console.log(JSON.stringify(summary)); diff --git a/frontend/src/app/api/consult/route.ts b/frontend/src/app/api/consult/route.ts index db0478b4..759d56f4 100644 --- a/frontend/src/app/api/consult/route.ts +++ b/frontend/src/app/api/consult/route.ts @@ -129,6 +129,12 @@ import { import { persistGuardedSessionTitle } from "@/lib/session-title-guard"; import { generateSessionTitle, shouldGenerateSessionTitle } from "@/lib/session-title-agent"; import { z } from "zod"; +import { CONSULTATION_QUEUE_FAILED_MESSAGE, isConsultationRateLimitFailure, retryOnConsultationRateLimit } from "@/lib/consultation-workflow-retry"; +import { + consumeUserRateLimitSet, + userRateLimitDeniedResponse, + userRateLimitMessages, +} from "@/lib/request-rate-limit"; export const runtime = "nodejs"; export const maxDuration = 480; @@ -367,6 +373,11 @@ export async function POST(request: Request) { const currentDate = chinaCalendarDate(requestTime); const userId = user.id; + const limited = consumeUserRateLimitSet(["consultMinute", "consultHour"], userId); + if (!limited.ok) { + const denied = userRateLimitDeniedResponse(limited.retryAfterSeconds, userRateLimitMessages.consult); + return NextResponse.json(denied.body, { status: denied.status, headers: denied.headers }); + } const requestId = parsed.data.requestId; const sessionId = parsed.data.sessionId; const consultationTheme = parsed.data.theme; @@ -1786,9 +1797,11 @@ export async function POST(request: Request) { theme: parsed.data.theme, }); const workflowContext = applyBirthTimeModeToWorkflowContext( - await runConsultationWorkflow(toolInput, { - foreground: true, - plan: prepared.preReserveResult, + await retryOnConsultationRateLimit({ + run: () => runConsultationWorkflow(toolInput, { + foreground: true, + plan: prepared.preReserveResult, + }), }), consultationMode, { birthTimeSource: prepared.serverChart.truth.birthTimeSource }, @@ -1894,12 +1907,15 @@ export async function POST(request: Request) { console.error( `[consult] generation failed request=${requestId} model=${selectedModel.id} reason=${reason}`, ); + const rateLimited = isConsultationRateLimitFailure(error); return NextResponse.json( - { - error: "暂时无法生成解读", - message: "咨询服务暂时不可用，请稍后再试。", - recovery: "稍后重试，或换一个模型继续。", - }, + rateLimited + ? { error: CONSULTATION_QUEUE_FAILED_MESSAGE, message: CONSULTATION_QUEUE_FAILED_MESSAGE } + : { + error: "暂时无法生成解读", + message: "咨询服务暂时不可用，请稍后再试。", + recovery: "稍后重试，或换一个模型继续。", + }, { status: 503 }, ); } diff --git a/frontend/src/app/api/rectification/agent/route.ts b/frontend/src/app/api/rectification/agent/route.ts index 08cf5d7a..35093fa3 100644 --- a/frontend/src/app/api/rectification/agent/route.ts +++ b/frontend/src/app/api/rectification/agent/route.ts @@ -18,6 +18,11 @@ import { replyToDeclaredBirthWindow, } from "@/lib/rectification-agentic/v9/agent-route-typed-message"; import { respondWithRectificationAgentTurn } from "@/lib/rectification-agentic/v9/agent-route-agent-turn"; +import { + consumeUserRateLimitSet, + userRateLimitDeniedResponse, + userRateLimitMessages, +} from "@/lib/request-rate-limit"; export const runtime = "nodejs"; export const maxDuration = 240; @@ -66,6 +71,20 @@ export async function POST(request: Request) { const frozen = await refuseWhenAccountDeletionPending(userId); if (frozen) return frozen; + // Opening a case is POST /api/rectification/cases/open and is not counted. + // Every turn on this route is, and the check is before any credit reserve. + const limited = consumeUserRateLimitSet( + ["rectificationTurnMinute", "rectificationTurnHour"], + userId, + ); + if (!limited.ok) { + const denied = userRateLimitDeniedResponse( + limited.retryAfterSeconds, + userRateLimitMessages.rectificationTurn, + ); + return NextResponse.json(denied.body, { status: denied.status, headers: denied.headers }); + } + if (action === "opening") { const consistency = await finalizeSuccessfulTurnExit({ accounting: accounting as never, userId, caseId, action, segmentOpening: true, diff --git a/frontend/src/app/api/reports/route.ts b/frontend/src/app/api/reports/route.ts index 3ca2e205..f3473d58 100644 --- a/frontend/src/app/api/reports/route.ts +++ b/frontend/src/app/api/reports/route.ts @@ -5,6 +5,7 @@ import { LONGFORM_SNAPSHOT_SECTION_ID } from "@/lib/personal-report-longform-sna import { resolveMissingBirthTimezoneOffset } from "@/lib/birth-profile-timezone"; import { stampReportBirthFingerprint } from "@/lib/report-birth-binding"; import { runConsultationWorkflow } from "@/mastra"; +import { consumeUserRateLimitSet, userRateLimitMessages } from "@/lib/request-rate-limit"; import { createPersonalReportAgent } from "@/mastra/personal-report"; import { loadLanguageModelCatalog } from "@/lib/model-catalog"; import { @@ -68,8 +69,8 @@ function sanitizedErrorCode(error: unknown): string { return "UnknownError"; } -function toNextResponse(response: { status: number; body: Record }) { - return NextResponse.json(response.body, { status: response.status }); +function toNextResponse(response: { status: number; body: Record; headers?: Readonly> }) { + return NextResponse.json(response.body, { status: response.status, headers: response.headers }); } function listReportView( @@ -308,6 +309,13 @@ export async function POST(request: Request) { }, persistence, model: defaultModel, + consumeReportRateLimit: userId + ? () => { + const decision = consumeUserRateLimitSet(["reportGenerateHour"], userId); + if (decision.ok) return decision; + return { ...decision, message: userRateLimitMessages.reportGenerate }; + } + : undefined, billing: { reserve: async ({ userId: billingUserId, requestId, modelId }) => { try { diff --git a/frontend/src/components/chat-message-row.tsx b/frontend/src/components/chat-message-row.tsx index afbb85fc..46e16984 100644 --- a/frontend/src/components/chat-message-row.tsx +++ b/frontend/src/components/chat-message-row.tsx @@ -61,6 +61,7 @@ export const ChatMessageRow = memo(function ChatMessageRow({ received: "working", "loading-method": "searching", "chart-calculation": "solving", + "compute-queue": "working", "evidence-validation": "working", "answer-composition": "composing", } as const)[message.activity.phase] diff --git a/frontend/src/lib/consultation-activity-labels.ts b/frontend/src/lib/consultation-activity-labels.ts index 1fcdf3c5..22fe90e1 100644 --- a/frontend/src/lib/consultation-activity-labels.ts +++ b/frontend/src/lib/consultation-activity-labels.ts @@ -4,6 +4,8 @@ * turn is classified or the chart is read (TASK-consult-first-frame-and-pacing-20260928 D1). */ export const CONSULTATION_RECEIVED_LABEL = "收到，正在看你的问题…"; +export const CONSULTATION_QUEUE_LABEL = "人有点多，排一下队"; +export const CONSULTATION_QUEUE_FAILED_MESSAGE = "人有点多，这一轮没排上，点数已经退回。"; export const CONSULTATION_LOADING_METHOD_LABEL = "正在读取印度占星分析规则…"; export const CONSULTATION_CHART_CALCULATION_LABEL = "正在计算本命盘…"; export const CONSULTATION_EVIDENCE_VALIDATION_LABEL = "正在核对可用证据…"; diff --git a/frontend/src/lib/consultation-agent-events.ts b/frontend/src/lib/consultation-agent-events.ts index de91456f..74a6e7d6 100644 --- a/frontend/src/lib/consultation-agent-events.ts +++ b/frontend/src/lib/consultation-agent-events.ts @@ -10,6 +10,9 @@ export const publicActivityPhaseSchema = z.enum([ "received", "loading-method", "chart-calculation", + // Heavy compute is full. The calculate row keeps this sentence while the + // consultation retries, then returns to the chart-calculation label. + "compute-queue", "evidence-validation", "answer-composition", ]); diff --git a/frontend/src/lib/consultation-run-timeline.ts b/frontend/src/lib/consultation-run-timeline.ts index 9dfc7092..3dc0dc2b 100644 --- a/frontend/src/lib/consultation-run-timeline.ts +++ b/frontend/src/lib/consultation-run-timeline.ts @@ -70,7 +70,7 @@ export function reduceConsultationTimeline( return applyCalculateHints(created); } if (event.type === "activity") { - if (event.phase === "chart-calculation") { + if (event.phase === "chart-calculation" || event.phase === "compute-queue") { return patchRow(state, CALCULATE_ID, { status: "live", label: event.label, diff --git a/frontend/src/lib/consultation-workflow-retry.ts b/frontend/src/lib/consultation-workflow-retry.ts new file mode 100644 index 00000000..2174db6f --- /dev/null +++ b/frontend/src/lib/consultation-workflow-retry.ts @@ -0,0 +1,57 @@ +import { ConsultationWorkflowError } from "../mastra/consultation-workflow.ts"; +import { + CONSULTATION_QUEUE_FAILED_MESSAGE, + CONSULTATION_QUEUE_LABEL, +} from "./consultation-activity-labels.ts"; + +export { CONSULTATION_QUEUE_FAILED_MESSAGE, CONSULTATION_QUEUE_LABEL }; + +export const CONSULTATION_QUEUE_MAX_RETRIES = 2; +export const CONSULTATION_QUEUE_MAX_WAIT_MS = 15_000; + +export function isConsultationRateLimitFailure(error: unknown): boolean { + if (error instanceof ConsultationWorkflowError) return error.code === "workflow_rate_limited"; + if (error instanceof Error && error.cause !== undefined) return isConsultationRateLimitFailure(error.cause); + return false; +} + +function retryDelaySeconds(error: unknown): number | undefined { + if (error instanceof ConsultationWorkflowError && error.code === "workflow_rate_limited") { + return error.retryAfterSeconds ?? 2; + } + if (error instanceof Error && error.cause !== undefined) return retryDelaySeconds(error.cause); + return undefined; +} + +export function consultationRunFailureMessage(error: unknown, fallback: string): string { + return isConsultationRateLimitFailure(error) ? CONSULTATION_QUEUE_FAILED_MESSAGE : fallback; +} + +/** + * Retry one consultation workflow call when the heavy-compute gate answers + * 429. At most two retries, and the sleeps together stay within 15 seconds. + */ +export async function retryOnConsultationRateLimit(input: Readonly<{ + run: () => Promise; + onQueue?: () => void | Promise; + sleep?: (ms: number) => Promise; +}>): Promise { + const sleep = input.sleep ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms))); + let retries = 0; + let waitedMs = 0; + for (;;) { + try { + return await input.run(); + } catch (error) { + const retryAfterSeconds = retryDelaySeconds(error); + if (retryAfterSeconds === undefined || retries >= CONSULTATION_QUEUE_MAX_RETRIES) throw error; + const remainingMs = CONSULTATION_QUEUE_MAX_WAIT_MS - waitedMs; + if (remainingMs <= 0) throw error; + const waitMs = Math.min(retryAfterSeconds * 1000, remainingMs); + retries += 1; + waitedMs += waitMs; + await input.onQueue?.(); + await sleep(waitMs); + } + } +} diff --git a/frontend/src/lib/personal-report-route-core.ts b/frontend/src/lib/personal-report-route-core.ts index 651efd77..a3a0e855 100644 --- a/frontend/src/lib/personal-report-route-core.ts +++ b/frontend/src/lib/personal-report-route-core.ts @@ -63,7 +63,11 @@ export const personalReportCreateRequestSchema = z.object({ export type PersonalReportCreateRequest = z.infer; -export type ReportRouteResponse = Readonly<{ status: number; body: Record }>; +export type ReportRouteResponse = Readonly<{ + status: number; + body: Record; + headers?: Readonly>; +}>; export type ReportServicePort = Pick< PersonalReportService, @@ -231,6 +235,10 @@ export type ReportCreateCoreDeps = Readonly<{ persistence: ReportServicePort; model: Readonly<{ id: string; configVersion?: number; inputCostMicrousdPerMillion?: number; outputCostMicrousdPerMillion?: number }> | null; billing?: ReportBillingPort; + /** Hourly per-user cap. Checked after a replay and the daily cap, before reserve. */ + consumeReportRateLimit?: () => + | { readonly ok: true } + | { readonly ok: false; readonly retryAfterSeconds: number; readonly message: string }; runWorkflow: (input: ConsultationInput) => Promise; createAgent: (model: Readonly<{ id: string }>) => ReportAgentPort; skillSnapshot: SkillSnapshot; @@ -387,6 +395,20 @@ export async function resolveReportCreate(deps: ReportCreateCoreDeps): Promise seenMaxWindowMs) seenMaxWindowMs = input.windowMs; - consumeCalls += 1; +function evaluateRequestRateLimit( + input: RequestRateLimitInput, + record: boolean, +): RequestRateLimitDecision { const now = input.now ?? Date.now(); const store = input.store ?? defaultStore(); - if (consumeCalls % 1_000 === 0 || store.size > 5_000) { - sweepExpiredRateLimitKeys(store, now, seenMaxWindowMs); + if (record) { + if (input.windowMs > seenMaxWindowMs) seenMaxWindowMs = input.windowMs; + consumeCalls += 1; + if (consumeCalls % 1_000 === 0 || store.size > 5_000) { + sweepExpiredRateLimitKeys(store, now, seenMaxWindowMs); + } } const windowStart = now - input.windowMs; const recent = (store.get(input.key) ?? []).filter((timestamp) => timestamp > windowStart); if (recent.length >= input.limit) { const retryAfterSeconds = Math.max(1, Math.ceil((recent[0]! + input.windowMs - now) / 1000)); - store.set(input.key, recent); + if (record) store.set(input.key, recent); return { ok: false, retryAfterSeconds }; } + if (!record) return { ok: true }; recent.push(now); store.set(input.key, recent); return { ok: true }; } +export function consumeRequestRateLimit(input: RequestRateLimitInput): RequestRateLimitDecision { + return evaluateRequestRateLimit(input, true); +} + export function consumeUserRequestRateLimit( bucket: RequestRateLimitBucket, userId: string, now?: number, + store?: RequestRateLimitStore, ): RequestRateLimitDecision { const spec = requestRateLimits[bucket]; return consumeRequestRateLimit({ @@ -85,5 +108,45 @@ export function consumeUserRequestRateLimit( limit: spec.limit, windowMs: spec.windowMs, now, + store, }); } + +/** + * Allow the request only when every bucket still has room. A refusal writes + * nothing, so the bucket that still had room is not charged for the miss. + */ +export function consumeUserRateLimitSet( + buckets: readonly RequestRateLimitBucket[], + userId: string, + now?: number, + store?: RequestRateLimitStore, +): RequestRateLimitDecision { + let retryAfterSeconds = 0; + for (const bucket of buckets) { + const spec = requestRateLimits[bucket]; + const decision = evaluateRequestRateLimit({ + key: `${bucket}:${userId}`, + limit: spec.limit, + windowMs: spec.windowMs, + now, + store, + }, false); + if (!decision.ok) retryAfterSeconds = Math.max(retryAfterSeconds, decision.retryAfterSeconds); + } + if (retryAfterSeconds > 0) return { ok: false, retryAfterSeconds }; + for (const bucket of buckets) consumeUserRequestRateLimit(bucket, userId, now, store); + return { ok: true }; +} + +export function userRateLimitDeniedResponse(retryAfterSeconds: number, message: string) { + return { + status: 429 as const, + headers: { "Retry-After": String(retryAfterSeconds) }, + body: { + error: message, + code: "rate_limited" as const, + retryAfterSeconds, + }, + }; +} diff --git a/frontend/src/lib/stream-agent-response.ts b/frontend/src/lib/stream-agent-response.ts index 6ed85b66..5401ecbd 100644 --- a/frontend/src/lib/stream-agent-response.ts +++ b/frontend/src/lib/stream-agent-response.ts @@ -22,6 +22,7 @@ import { } from "./consultation-step-timing.ts"; import { createVisibleTextTransformer } from "./stream-text-response.ts"; import { consultationWriteLabel } from "./consultation-activity-labels.ts"; +import { consultationRunFailureMessage } from "./consultation-workflow-retry.ts"; import { logTruncatedReasoning } from "./consultation-budget.ts"; import { classifyPass4, @@ -1177,7 +1178,7 @@ export function streamAgentResponse(options: StreamAgentResponseOptions) { send(controller, { type: "run.failed", code, - message: runFailedMessage(code), + message: consultationRunFailureMessage(error, runFailedMessage(code)), ...(failureReceipt ? { receipt: failureReceipt } : {}), }); if (!disconnected) controller.close(); diff --git a/frontend/src/mastra/consultation-tools.ts b/frontend/src/mastra/consultation-tools.ts index c5c3fd8f..b2ad3868 100644 --- a/frontend/src/mastra/consultation-tools.ts +++ b/frontend/src/mastra/consultation-tools.ts @@ -27,6 +27,7 @@ import { normalizeTechniqueAuditRows } from "../lib/consultation-technique-audit import type { AgentModelFinishReason, EvidenceCardTelemetry } from "../lib/agent-observability.ts"; import { agentGenerationSettings, AGENT_SLICE_ANSWER_OUTPUT_TOKENS, AGENT_SLICE_THINKING_OUTPUT_TOKENS, type ProviderReasoningEffort } from "../lib/agent-generation-settings.ts"; import { chartCalculationProgressLabel, evidenceLookupActivityLabel } from "../lib/consultation-activity-labels.ts"; +import { CONSULTATION_QUEUE_LABEL, retryOnConsultationRateLimit } from "../lib/consultation-workflow-retry.ts"; import { consultDomainConcurrency } from "../lib/consult/prep-flags.ts"; import { buildEvidenceCard, @@ -1309,6 +1310,36 @@ export function createConsultationTools(ctx: ConsultationAgentContext) { try { const userIntent = ctx.plan?.userIntent ?? input.question; const executions: DomainExecution[] = []; + const runQueuedWorkflow = async ( + toolInput: ReturnType, + domainPlan: ConsultationPlan, + progressLabel: string, + ) => { + let queued = false; + const workflow = await retryOnConsultationRateLimit({ + run: () => (ctx.runWorkflow ?? runConsultationWorkflow)(toolInput, { + foreground: true, + signal: context.abortSignal ?? ctx.abortSignal, + plan: domainPlan, + requestId: ctx.requestId, + entrypoint: ctx.entrypoint, + }), + onQueue: async () => { + queued = true; + await context.writer?.custom({ + type: "data-jyotish-activity", + data: { phase: "compute-queue", label: CONSULTATION_QUEUE_LABEL }, + }); + }, + }); + if (queued) { + await context.writer?.custom({ + type: "data-jyotish-activity", + data: { phase: "chart-calculation", label: progressLabel }, + }); + } + return workflow; + }; if (consultDomainConcurrency() === 1) { for (let index = 0; index < domains.length; index += 1) { const domain = domains[index]; @@ -1344,13 +1375,11 @@ export function createConsultationTools(ctx: ConsultationAgentContext) { question: userIntent, theme: domain, }); - const workflow = await (ctx.runWorkflow ?? runConsultationWorkflow)(toolInput, { - foreground: true, - signal: context.abortSignal ?? ctx.abortSignal, - plan: domainPlan, - requestId: ctx.requestId, - entrypoint: ctx.entrypoint, - }); + const workflow = await runQueuedWorkflow( + toolInput, + domainPlan, + chartCalculationProgressLabel(index + 1, domains.length), + ); const guarded = applyBirthTimeModeToWorkflowContext(workflow, ctx.consultationMode, { birthTimeSource: ctx.serverChart.truth.birthTimeSource, }); @@ -1374,7 +1403,7 @@ export function createConsultationTools(ctx: ConsultationAgentContext) { }, }); } - const parallelExecutions = await Promise.all(domains.map(async (domain) => { + const parallelExecutions = await Promise.all(domains.map(async (domain, domainIndex) => { if (!domain) return null; const domainPlan = ctx.plan && domains.length === 1 @@ -1396,13 +1425,11 @@ export function createConsultationTools(ctx: ConsultationAgentContext) { question: userIntent, theme: domain, }); - const workflow = await (ctx.runWorkflow ?? runConsultationWorkflow)(toolInput, { - foreground: true, - signal: context.abortSignal ?? ctx.abortSignal, - plan: domainPlan, - requestId: ctx.requestId, - entrypoint: ctx.entrypoint, - }); + const workflow = await runQueuedWorkflow( + toolInput, + domainPlan, + chartCalculationProgressLabel(domainIndex + 1, domains.length), + ); const guarded = applyBirthTimeModeToWorkflowContext(workflow, ctx.consultationMode, { birthTimeSource: ctx.serverChart.truth.birthTimeSource, }); diff --git a/frontend/src/mastra/consultation-workflow.ts b/frontend/src/mastra/consultation-workflow.ts index bb44d4b8..e1cad12d 100644 --- a/frontend/src/mastra/consultation-workflow.ts +++ b/frontend/src/mastra/consultation-workflow.ts @@ -102,14 +102,23 @@ export type ConsultationWorkflowFailureCode = typeof consultationWorkflowFailure export class ConsultationWorkflowError extends Error { readonly code: ConsultationWorkflowFailureCode; + readonly retryAfterSeconds?: number; - constructor(code: ConsultationWorkflowFailureCode, message: string) { + constructor(code: ConsultationWorkflowFailureCode, message: string, retryAfterSeconds?: number) { super(message); this.name = "ConsultationWorkflowError"; this.code = code; + if (retryAfterSeconds !== undefined) this.retryAfterSeconds = retryAfterSeconds; } } +function parseRetryAfterHeader(raw: string | null): number { + if (!raw) return 2; + const seconds = Number(raw.trim()); + if (!Number.isFinite(seconds)) return 2; + return Math.max(1, Math.ceil(seconds)); +} + function httpFailureCode(status: number): ConsultationWorkflowFailureCode { if (status === 400) return "workflow_bad_request"; if (status === 403) return "workflow_forbidden"; @@ -174,6 +183,7 @@ export async function runConsultationWorkflow( throw new ConsultationWorkflowError( response.ok ? "workflow_empty_response" : httpFailureCode(response.status), data?.error || data?.message || `Jyotish API returned ${response.status}`, + response.status === 429 ? parseRetryAfterHeader(response.headers.get("retry-after")) : undefined, ); } const parsed = consultationWorkflowResponseSchema.safeParse(data); diff --git a/frontend/tests/consultation-workflow-retry.test.ts b/frontend/tests/consultation-workflow-retry.test.ts new file mode 100644 index 00000000..63cfb405 --- /dev/null +++ b/frontend/tests/consultation-workflow-retry.test.ts @@ -0,0 +1,102 @@ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import test from "node:test"; +import { ConsultationWorkflowError } from "../src/mastra/consultation-workflow.ts"; +import { CONSULTATION_QUEUE_LABEL } from "../src/lib/consultation-activity-labels.ts"; +import { + CONSULTATION_QUEUE_FAILED_MESSAGE, + consultationRunFailureMessage, + isConsultationRateLimitFailure, + retryOnConsultationRateLimit, +} from "../src/lib/consultation-workflow-retry.ts"; +import { + emptyConsultationTimeline, + reduceConsultationTimelineEvents, +} from "../src/lib/consultation-run-timeline.ts"; + +test("a 429 is retried once and the queue line is shown before success", async () => { + let calls = 0; + const queued: string[] = []; + const sleeps: number[] = []; + const result = await retryOnConsultationRateLimit({ + run: async () => { + calls += 1; + if (calls === 1) throw new ConsultationWorkflowError("workflow_rate_limited", "busy", 2); + return "ok"; + }, + onQueue: () => { queued.push(CONSULTATION_QUEUE_LABEL); }, + sleep: async (ms) => { sleeps.push(ms); }, + }); + assert.equal(result, "ok"); + assert.equal(calls, 2); + assert.deepEqual(sleeps, [2000]); + assert.deepEqual(queued, ["人有点多，排一下队"]); +}); + +test("three 429s stop within two retries and 15 seconds", async () => { + let calls = 0; + let slept = 0; + await assert.rejects( + () => retryOnConsultationRateLimit({ + run: async () => { + calls += 1; + throw new ConsultationWorkflowError("workflow_rate_limited", "busy", 2); + }, + sleep: async (ms) => { slept += ms; }, + }), + (error: unknown) => { + assert.equal(isConsultationRateLimitFailure(error), true); + return true; + }, + ); + assert.equal(calls, 3); + assert.equal(slept, 4_000); + assert.ok(slept <= 15_000); + assert.equal( + consultationRunFailureMessage(new ConsultationWorkflowError("workflow_rate_limited", "busy", 2), "其他失败"), + CONSULTATION_QUEUE_FAILED_MESSAGE, + ); + assert.equal(consultationRunFailureMessage(new Error("other"), "其他失败"), "其他失败"); +}); + +test("Retry-After cannot push the total wait past 15 seconds", async () => { + let calls = 0; + let slept = 0; + await assert.rejects(() => retryOnConsultationRateLimit({ + run: async () => { + calls += 1; + throw new ConsultationWorkflowError("workflow_rate_limited", "busy", 20); + }, + sleep: async (ms) => { slept += ms; }, + })); + assert.equal(calls, 2); + assert.equal(slept, 15_000); +}); + +test("the queue sentence replaces the calculate row and the chart label comes back", () => { + const calculating = reduceConsultationTimelineEvents([ + { type: "tool.started", callId: "tool-1", tool: "run-jyotish-consultation", label: "正在计算个人星盘" }, + { type: "activity", phase: "chart-calculation", label: "正在计算本命盘…" }, + ], emptyConsultationTimeline()); + const queued = reduceConsultationTimelineEvents([ + { type: "activity", phase: "compute-queue", label: CONSULTATION_QUEUE_LABEL }, + ], calculating); + assert.equal(queued.rows.find((row) => row.kind === "calculate")?.label, "人有点多，排一下队"); + assert.equal(queued.rows.find((row) => row.kind === "calculate")?.status, "live"); + const resumed = reduceConsultationTimelineEvents([ + { type: "activity", phase: "chart-calculation", label: "正在计算本命盘…" }, + ], queued); + assert.equal(resumed.rows.find((row) => row.kind === "calculate")?.label, "正在计算本命盘…"); +}); + +test("a failed consultation still settles through the existing cancel path before the queue sentence", () => { + const stream = readFileSync(new URL("../src/lib/stream-agent-response.ts", import.meta.url), "utf8"); + const consult = readFileSync(new URL("../src/app/api/consult/route.ts", import.meta.url), "utf8"); + const onErrorAt = stream.indexOf("await options.onError?."); + const messageAt = stream.indexOf("consultationRunFailureMessage(error, runFailedMessage(code))"); + assert.ok(onErrorAt > 0 && messageAt > onErrorAt); + assert.match(consult, /onError: \(error\) => settleRun\(\s*cancel/); + assert.match(consult, /else await cancel\(\)/); + assert.match(consult, /retryOnConsultationRateLimit/); + assert.match(consult, /CONSULTATION_QUEUE_FAILED_MESSAGE/); +}); diff --git a/frontend/tests/personal-report-api.test.ts b/frontend/tests/personal-report-api.test.ts index 0470ea2a..af7676d5 100644 --- a/frontend/tests/personal-report-api.test.ts +++ b/frontend/tests/personal-report-api.test.ts @@ -710,6 +710,19 @@ test("core create: 429 at the daily limit", async () => { assert.equal(response.body.code, "report_rate_limited"); }); +test("core create: hourly rate limit returns 429 and does not reserve", async () => { + const billing = billingHarness(); + const response = await resolveReportCreate(baseDeps({ + billing: billing.billing, + consumeReportRateLimit: () => ({ ok: false, retryAfterSeconds: 12, message: "这一小时的报告已经够多了，先停一下。" }), + })); + assert.equal(response.status, 429); + assert.equal(response.body.code, "rate_limited"); + assert.equal(response.body.retryAfterSeconds, 12); + assert.equal(response.headers?.["Retry-After"], "12"); + assert.equal(billing.reservations, 0); +}); + test("core create: 502 model_unavailable when no model is configured", async () => { const persistence = new MemoryPersistence(); const response = await resolveReportCreate(baseDeps({ diff --git a/frontend/tests/request-rate-limit.test.ts b/frontend/tests/request-rate-limit.test.ts index d9c30d29..a37d0119 100644 --- a/frontend/tests/request-rate-limit.test.ts +++ b/frontend/tests/request-rate-limit.test.ts @@ -1,8 +1,13 @@ import assert from "node:assert/strict"; import test from "node:test"; +import { readFileSync } from "node:fs"; import { consumeRequestRateLimit, + consumeUserRateLimitSet, + requestRateLimits, resetRequestRateLimitMaintenanceForTests, + userRateLimitDeniedResponse, + userRateLimitMessages, } from "../src/lib/request-rate-limit.ts"; test("a user is allowed up to the window limit and then must wait", () => { @@ -110,3 +115,98 @@ test("keys past the longest seen window are swept, and a one-hour key survives 6 assert.equal(mixed.has("ephemeris:person"), true); resetRequestRateLimitMaintenanceForTests(); }); + +test("consult allows 10 per minute and refuses the 11th without charging the hour bucket further", () => { + const store = new Map(); + for (let index = 0; index < 10; index += 1) { + const decision = consumeUserRateLimitSet( + ["consultMinute", "consultHour"], + "person", + 1_000 + index, + store, + ); + assert.deepEqual(decision, { ok: true }); + } + const denied = consumeUserRateLimitSet(["consultMinute", "consultHour"], "person", 2_000, store); + assert.equal(denied.ok, false); + if (!denied.ok) assert.equal(denied.retryAfterSeconds, 59); + assert.equal(store.get("consultMinute:person")?.length, 10); + assert.equal(store.get("consultHour:person")?.length, 10); + assert.equal(requestRateLimits.consultMinute.limit, 10); + assert.equal(requestRateLimits.consultHour.limit, 120); +}); + +test("the hour bucket can refuse while the minute bucket still has room, and neither bucket records the miss", () => { + const store = new Map(); + for (let index = 0; index < 120; index += 1) { + consumeRequestRateLimit({ + key: "consultHour:person", + limit: 120, + windowMs: 60 * 60_000, + now: index * 1_000, + store, + }); + } + const denied = consumeUserRateLimitSet(["consultMinute", "consultHour"], "person", 130_000, store); + assert.equal(denied.ok, false); + assert.equal(store.has("consultMinute:person"), false); + assert.equal(store.get("consultHour:person")?.length, 120); +}); + +test("rectification turns allow 5 per minute and 40 per hour, reports allow 3 per hour", () => { + const store = new Map(); + for (let index = 0; index < 5; index += 1) { + assert.deepEqual( + consumeUserRateLimitSet(["rectificationTurnMinute", "rectificationTurnHour"], "person", 5_000 + index, store), + { ok: true }, + ); + } + const denied = consumeUserRateLimitSet( + ["rectificationTurnMinute", "rectificationTurnHour"], + "person", + 5_100, + store, + ); + assert.equal(denied.ok, false); + assert.equal(store.get("rectificationTurnMinute:person")?.length, 5); + assert.equal(requestRateLimits.rectificationTurnMinute.limit, 5); + assert.equal(requestRateLimits.rectificationTurnHour.limit, 40); + assert.equal(requestRateLimits.reportGenerateHour.limit, 3); + assert.equal(requestRateLimits.reportGenerateHour.windowMs, 60 * 60_000); + + const reports = new Map(); + for (let index = 0; index < 3; index += 1) { + assert.deepEqual( + consumeUserRateLimitSet(["reportGenerateHour"], "person", 8_000 + index, reports), + { ok: true }, + ); + } + const reportDenied = consumeUserRateLimitSet(["reportGenerateHour"], "person", 9_000, reports); + assert.equal(reportDenied.ok, false); + assert.equal(reports.get("reportGenerateHour:person")?.length, 3); +}); + +test("a denied response carries 429, rate_limited, and Retry-After", () => { + const denied = userRateLimitDeniedResponse(12, userRateLimitMessages.consult); + assert.equal(denied.status, 429); + assert.equal(denied.headers["Retry-After"], "12"); + assert.deepEqual(denied.body, { + error: "这一会儿问得有点密，先停一下。", + code: "rate_limited", + retryAfterSeconds: 12, + }); +}); + +test("consult, rectification turns, and report creation check the limit before they reserve", () => { + const consult = readFileSync(new URL("../src/app/api/consult/route.ts", import.meta.url), "utf8"); + const rectification = readFileSync(new URL("../src/app/api/rectification/agent/route.ts", import.meta.url), "utf8"); + const caseOpen = readFileSync(new URL("../src/app/api/rectification/cases/open/route.ts", import.meta.url), "utf8"); + const reports = readFileSync(new URL("../src/app/api/reports/route.ts", import.meta.url), "utf8"); + const reportCore = readFileSync(new URL("../src/lib/personal-report-route-core.ts", import.meta.url), "utf8"); + assert.ok(consult.indexOf("consumeUserRateLimitSet(") < consult.indexOf("reserve_consultation_usage")); + assert.match(consult, /userRateLimitDeniedResponse\(/); + assert.ok(rectification.indexOf("consumeUserRateLimitSet(") < rectification.indexOf("respondWithRectificationAgentTurn(")); + assert.equal(caseOpen.includes("consumeUserRateLimitSet"), false); + assert.match(reports, /consumeUserRateLimitSet\(\["reportGenerateHour"\]/); + assert.ok(reportCore.indexOf("if (deps.consumeReportRateLimit)") < reportCore.indexOf("deps.billing.reserve")); +}); diff --git a/scripts/api_heavy_compute_gate.py b/scripts/api_heavy_compute_gate.py index 0067a9b9..5789cc6f 100644 --- a/scripts/api_heavy_compute_gate.py +++ b/scripts/api_heavy_compute_gate.py @@ -1,9 +1,11 @@ -"""Fail-fast bounded concurrency for heavy Jyotish API compute endpoints. +"""Bounded concurrency for heavy Jyotish API compute endpoints. -Request threads that run rectification scans or high-rigor workflows share one -process-wide semaphore sized for the 2 vCPU production host. Saturated requests -return immediately; they are not queued. Health checks and other light routes -must not call this gate. +Request threads that run rectification scans or consultation workflows share one +process-wide semaphore sized for the 2 vCPU production host. A saturated request +waits up to 8 seconds when fewer than 4 threads are already waiting. If that +queue is full or the wait expires, the caller still gets HTTP 429 and +Retry-After. A thread is never parked without a deadline. Health checks and +other light routes must not call this gate. """ from __future__ import annotations @@ -15,6 +17,8 @@ import time DEFAULT_CONCURRENCY = 2 DEFAULT_RETRY_AFTER_SECONDS = 2 +DEFAULT_QUEUE_WAIT_SECONDS = 8.0 +DEFAULT_MAX_WAITERS = 4 ENV_CONCURRENCY = "JYOTISH_HEAVY_COMPUTE_CONCURRENCY" ENV_RETRY_AFTER = "JYOTISH_HEAVY_COMPUTE_RETRY_AFTER_SECONDS" ENV_GATE_CHART_OVERVIEW = "JYOTISH_GATE_CHART_OVERVIEW" @@ -51,6 +55,9 @@ _state_lock = threading.Lock() _semaphore: threading.BoundedSemaphore | None = None _retry_after_seconds = DEFAULT_RETRY_AFTER_SECONDS _limit = DEFAULT_CONCURRENCY +_wait_seconds = DEFAULT_QUEUE_WAIT_SECONDS +_max_waiters = DEFAULT_MAX_WAITERS +_waiters = 0 _holders: list[dict] = [] _holder_seq = 0 @@ -73,7 +80,7 @@ class _HeldSlot: class HeavyComputeBusy(RuntimeError): - """No heavy-compute slot is free; callers must fail fast with HTTP 429.""" + """The short queue could not take this request; callers return HTTP 429.""" error_code = "ERR_COMPUTE_BUSY" @@ -102,9 +109,11 @@ def reset_heavy_compute_gate( *, limit: int | None = None, retry_after_seconds: int | None = None, + wait_seconds: float | None = None, + max_waiters: int | None = None, ) -> None: """Rebuild the process-wide semaphore. Tests must call this after env changes.""" - global _semaphore, _retry_after_seconds, _limit, _holders + global _semaphore, _retry_after_seconds, _limit, _holders, _wait_seconds, _max_waiters, _waiters resolved_limit = ( limit if limit is not None @@ -115,9 +124,14 @@ def reset_heavy_compute_gate( if retry_after_seconds is not None else _parse_positive_int(os.environ.get(ENV_RETRY_AFTER), DEFAULT_RETRY_AFTER_SECONDS) ) + resolved_wait = DEFAULT_QUEUE_WAIT_SECONDS if wait_seconds is None else max(0.0, float(wait_seconds)) + resolved_waiters = DEFAULT_MAX_WAITERS if max_waiters is None else max(0, int(max_waiters)) with _state_lock: _retry_after_seconds = resolved_retry _limit = resolved_limit + _wait_seconds = resolved_wait + _max_waiters = resolved_waiters + _waiters = 0 _holders = [] _semaphore = threading.BoundedSemaphore(resolved_limit) @@ -143,35 +157,61 @@ def _holder_view(now: float) -> list[dict]: ] -def _acquire_recorded(path: str) -> _HeldSlot: - """Take one permit and remember which path holds it. Fail fast when full.""" +def _remember_locked(path: str, semaphore: threading.BoundedSemaphore) -> _HeldSlot: + """Record a permit that this thread already holds. Caller owns ``_state_lock``.""" global _holder_seq + _holder_seq += 1 + holder_id = _holder_seq + _holders.append({"id": holder_id, "path": path, "started": time.monotonic()}) + return _HeldSlot(semaphore, holder_id) + + +def _reject_busy(path: str, retry_after: int) -> None: + with _state_lock: + snapshot = _holder_view(time.monotonic()) + _LOG.warning( + "heavy_compute_busy rejected_path=%s holders=%s", + path, + snapshot, + ) + raise HeavyComputeBusy(retry_after) + + +def _acquire_recorded(path: str) -> _HeldSlot: + """Take one permit. Wait briefly when full; still raise once the queue cannot.""" + global _waiters with _state_lock: semaphore, retry_after = _ensure_locked() - if not semaphore.acquire(blocking=False): - snapshot = _holder_view(time.monotonic()) - failed = True - slot = None + if semaphore.acquire(blocking=False): + return _remember_locked(path, semaphore) + if _waiters >= _max_waiters: + queued = False else: - _holder_seq += 1 - holder_id = _holder_seq - _holders.append({"id": holder_id, "path": path, "started": time.monotonic()}) - failed = False - snapshot = [] - slot = _HeldSlot(semaphore, holder_id) - if failed: - _LOG.warning( - "heavy_compute_busy rejected_path=%s holders=%s", - path, - snapshot, - ) - raise HeavyComputeBusy(retry_after) - assert slot is not None - return slot + _waiters += 1 + queued = True + wait_seconds = _wait_seconds + if not queued: + _reject_busy(path, retry_after) + acquired = False + try: + acquired = bool(semaphore.acquire(blocking=True, timeout=wait_seconds)) + finally: + with _state_lock: + _waiters -= 1 + if not acquired: + _reject_busy(path, retry_after) + with _state_lock: + return _remember_locked(path, semaphore) + + +def heavy_compute_waiter_count() -> int: + """Threads parked inside the short queue. Tests use this to avoid races.""" + with _state_lock: + return _waiters def acquire_heavy_compute_slot(path: str) -> _HeldSlot | None: - """Acquire a slot for a gated path. Light paths return None. Fail-fast on saturation.""" + """Acquire a slot for a gated path. Light paths return None.""" if not is_heavy_compute_path(path): return None return _acquire_recorded(path) diff --git a/tests/test_api_chart_subprocess_gate.py b/tests/test_api_chart_subprocess_gate.py index e8a97821..1e1ae8cb 100644 --- a/tests/test_api_chart_subprocess_gate.py +++ b/tests/test_api_chart_subprocess_gate.py @@ -153,7 +153,10 @@ def test_chart_overview_gate_is_off_by_default(monkeypatch) -> None: def test_chart_overview_gate_only_sync_unskipped_requests(monkeypatch) -> None: monkeypatch.setenv(gate.ENV_GATE_CHART_OVERVIEW, "1") - gate.reset_heavy_compute_gate(limit=1) + # 原值：reset(limit=1) 后第二次立刻 HeavyComputeBusy + # 新值：wait_seconds=0 + # 原因：默认改为短排队；本测试仍锁「名额被占时同步排盘被拒绝」，等待行为不在这里 + gate.reset_heavy_compute_gate(limit=1, wait_seconds=0) body = {"year": 1990, "lat": 28.6, "lon": 77.2} slot = gate.acquire_heavy_compute_slot_for_request("/api/chart", body) assert slot is not None @@ -172,7 +175,10 @@ def test_chart_overview_gate_only_sync_unskipped_requests(monkeypatch) -> None: def test_busy_log_names_holders_and_not_the_body(caplog) -> None: - gate.reset_heavy_compute_gate(limit=1) + # 原值：reset(limit=1) 后拒绝日志立刻出现 + # 新值：wait_seconds=0 + # 原因：默认会先等 8 秒；本测试仍锁日志只有路径、没有请求正文 + gate.reset_heavy_compute_gate(limit=1, wait_seconds=0) held = gate.acquire_heavy_compute_slot("/api/professional_report_reference") assert held is not None secret_body = {"birth_date": "1990-01-01", "name": "must-not-appear"} diff --git a/tests/test_api_heavy_compute_gate.py b/tests/test_api_heavy_compute_gate.py index 40195778..8481bc07 100644 --- a/tests/test_api_heavy_compute_gate.py +++ b/tests/test_api_heavy_compute_gate.py @@ -4,6 +4,7 @@ from __future__ import annotations import json import threading +import time from io import BytesIO import pytest @@ -14,6 +15,7 @@ from scripts.api_heavy_compute_gate import ( HEAVY_COMPUTE_PATHS, HeavyComputeBusy, acquire_heavy_compute_slot, + heavy_compute_waiter_count, is_heavy_compute_path, release_heavy_compute_slot, reset_heavy_compute_gate, @@ -93,7 +95,10 @@ def limit_one_gate(monkeypatch): monkeypatch.setenv("JYOTISH_HEAVY_COMPUTE_CONCURRENCY", "1") monkeypatch.setenv("JYOTISH_HEAVY_COMPUTE_RETRY_AFTER_SECONDS", "2") monkeypatch.setenv("JYOTISH_API_RATE_LIMIT_PER_MINUTE", "0") - reset_heavy_compute_gate() + # 原值：reset() 使用默认「满了立刻拒绝」 + # 新值：wait_seconds=0，本夹具里的 429 断言仍是立刻拒绝 + # 原因：产品改为最多等 8 秒；排队成功、等待者满、超时另有测试 + reset_heavy_compute_gate(wait_seconds=0) yield monkeypatch.delenv("JYOTISH_HEAVY_COMPUTE_CONCURRENCY", raising=False) monkeypatch.delenv("JYOTISH_HEAVY_COMPUTE_RETRY_AFTER_SECONDS", raising=False) @@ -119,7 +124,10 @@ def test_health_and_light_paths_are_not_gated() -> None: def test_acquire_fail_fast_then_succeeds_after_release() -> None: - reset_heavy_compute_gate(limit=1, retry_after_seconds=3) + # 原值：reset(limit=1, retry_after_seconds=3) 后第二路立刻 HeavyComputeBusy + # 新值：多传 wait_seconds=0 + # 原因：默认改为短排队；本测试仍锁 Retry-After=3 与 ERR_COMPUTE_BUSY，不锁「禁止等待」 + reset_heavy_compute_gate(limit=1, retry_after_seconds=3, wait_seconds=0) first = acquire_heavy_compute_slot("/api/high_rigor_workflow") assert first is not None with pytest.raises(HeavyComputeBusy) as caught: @@ -248,8 +256,87 @@ def test_warm_consultation_does_not_consume_the_user_slot(limit_one_gate, monkey assert "packet" not in warm_box["payload"] +def _wait_until(predicate, timeout: float = 2.0) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.01) + return False + + +def test_third_request_gets_the_slot_after_one_holder_releases() -> None: + reset_heavy_compute_gate(limit=2, retry_after_seconds=2, wait_seconds=2, max_waiters=4) + first = acquire_heavy_compute_slot("/api/consultation_workflow") + second = acquire_heavy_compute_slot("/api/rectification/v5/score") + assert first is not None and second is not None + found: dict = {} + + def _third() -> None: + found["slot"] = acquire_heavy_compute_slot("/api/consultation_workflow") + + worker = threading.Thread(target=_third) + worker.start() + assert _wait_until(lambda: heavy_compute_waiter_count() == 1) + release_heavy_compute_slot(first) + worker.join(timeout=3) + assert not worker.is_alive() + assert found["slot"] is not None + release_heavy_compute_slot(second) + release_heavy_compute_slot(found["slot"]) + reset_heavy_compute_gate() + + +def test_waiter_overflow_rejects_immediately() -> None: + reset_heavy_compute_gate(limit=1, retry_after_seconds=2, wait_seconds=2, max_waiters=1) + held = acquire_heavy_compute_slot("/api/high_rigor_workflow") + assert held is not None + parked: dict = {} + + def _wait() -> None: + try: + parked["slot"] = acquire_heavy_compute_slot("/api/consultation_workflow") + except HeavyComputeBusy as exc: + parked["error"] = exc + + waiter = threading.Thread(target=_wait) + waiter.start() + assert _wait_until(lambda: heavy_compute_waiter_count() == 1) + started = time.monotonic() + with pytest.raises(HeavyComputeBusy) as caught: + acquire_heavy_compute_slot("/api/rectification/v5/score") + assert time.monotonic() - started < 0.3 + assert caught.value.retry_after_seconds == 2 + assert caught.value.error_code == "ERR_COMPUTE_BUSY" + release_heavy_compute_slot(held) + waiter.join(timeout=3) + assert not waiter.is_alive() + assert parked.get("slot") is not None + release_heavy_compute_slot(parked["slot"]) + reset_heavy_compute_gate() + + +def test_queue_timeout_still_returns_busy() -> None: + reset_heavy_compute_gate(limit=1, retry_after_seconds=2, wait_seconds=0.2, max_waiters=4) + held = acquire_heavy_compute_slot("/api/professional_report_reference") + assert held is not None + started = time.monotonic() + with pytest.raises(HeavyComputeBusy) as caught: + acquire_heavy_compute_slot("/api/consultation_workflow") + elapsed = time.monotonic() - started + assert 0.1 <= elapsed < 2.0 + assert caught.value.retry_after_seconds == 2 + assert caught.value.error_code == "ERR_COMPUTE_BUSY" + light = acquire_heavy_compute_slot("/api/health") + assert light is None + release_heavy_compute_slot(held) + reset_heavy_compute_gate() + + def test_quality_gate_runs_heavy_compute_gate() -> None: from scripts.run_quality_gate import CORE_PYTEST_TARGETS assert "tests/test_api_heavy_compute_gate.py" in CORE_PYTEST_TARGETS assert gate.DEFAULT_CONCURRENCY == 2 + assert gate.DEFAULT_QUEUE_WAIT_SECONDS == 8 + assert gate.DEFAULT_MAX_WAITERS == 4