From f748dc72c3e30c45b777bff881445cfb8b165aa2 Mon Sep 17 00:00:00 2001 From: kdletters Date: Mon, 20 Jul 2026 20:56:17 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=B6=E5=8F=A3=E5=9B=BE=E7=89=87=E7=94=9F?= =?UTF-8?q?=E6=88=90=E4=BB=BB=E5=8A=A1=E8=B6=85=E6=97=B6=E9=A2=84=E7=AE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 将四类 VectorEngine 图片任务纳入长任务预算并预留终态写回窗口 - 通过请求上下文传递绝对截止时间并约束发送重试与图片下载 - 允许显式下调单次请求超时并补齐预算回归测试 - 同步外部生成 Worker 运维文档与项目记忆 --- .../shared-memory/decision-log.md | 4 + docs/project-memory/shared-memory/pitfalls.md | 6 + ...端架构】外部生成Worker化方案-2026-06-03.md | 8 +- ...发运维】本地开发验证与生产运维-2026-05-15.md | 6 +- server-rs/crates/api-server/src/config.rs | 15 +- .../api-server/src/edutainment_baby_object.rs | 1 + .../src/external_generation_worker.rs | 171 ++++++++++++--- server-rs/crates/api-server/src/jump_hop.rs | 2 + .../api-server/src/openai_image_generation.rs | 40 ++++ .../api-server/src/puzzle/vector_engine.rs | 4 + .../crates/api-server/src/request_context.rs | 28 +++ .../src/vector_engine/budget.rs | 179 +++++++++++++++ .../src/vector_engine/client.rs | 205 +++++++++++++++--- .../src/vector_engine/image_source.rs | 175 +++++++++++++-- .../platform-image/src/vector_engine/mod.rs | 1 + .../src/vector_engine/response.rs | 22 +- .../platform-image/src/vector_engine/tests.rs | 2 + .../platform-image/src/vector_engine/types.rs | 1 + .../platform-image/tests/vector_engine.rs | 72 +++++- 19 files changed, 846 insertions(+), 96 deletions(-) create mode 100644 server-rs/crates/platform-image/src/vector_engine/budget.rs diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index 57a8898bf..5d05df01a 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -4340,3 +4340,7 @@ - Vite 全量边界补充：`src/games/**`、`src/data/**`、`src/prompts/**`、旧顶层 App / Playground、旧路由和 `services/ai.ts` 必须由 pre-transform 门禁直接拒绝；所有同源 `/generated-*` 裸读在 dev 与生产统一为空 `404`，历史对象只经现役签名读取接口兼容，不允许 SPA fallback 伪装成资产成功响应。 - 前端混合根目录补充：`src/components`、`src/hooks`、`src/persistence`、`src/routing`、`src/services` 的根级文件实行现役白名单，Vite 与 ESLint 使用同一口径阻断旧 RPG / 玩法根文件；子目录仍按现役目录和退役目录分别管理，新增公共根文件必须显式登记。 - 影响范围：`PlatformEntryActiveFlowShell`、`PlatformActiveProfileView`、编辑器 / 项目搜索、平台 profile clients、`module-runtime`、`platform-llm`、Cargo workspace / resolve graph、Vite / ESLint / Rust 产物门禁及旧业务退役方案。 + +## 2026-07-20 VectorEngine 图片任务预算收口到 worker deadline + +- 决策：`editor_image_generation`、`editor_image_edit`、`editor_icon_spritesheet_generation` 和 `editor_ui_design_asset_extraction` 使用默认 `1800s` long job 预算。worker 从同一起点计算绝对 job deadline，并向 provider 提前保留 `min(60s, job 预算 / 2)` 作为审计、OSS 和终态写回窗口。deadline 只经进程内 `RequestContext` 传递；VectorEngine 单 attempt 取配置 timeout 与剩余预算的较小值，退避加下一次 attempt 无法落在同一 deadline 内时停止重试，参考图和响应图片下载也受同一 deadline 限制。普通 HTTP / `inline` 保持无 deadline 行为；`VECTOR_ENGINE_IMAGE_REQUEST_TIMEOUT_MS` 默认仍为 `1000000`，配置加载层允许显式值更低。lease 续租 / fencing、迟到写回仲裁、attempt 耗尽和原子退款语义不变。 diff --git a/docs/project-memory/shared-memory/pitfalls.md b/docs/project-memory/shared-memory/pitfalls.md index d2ed6fe6d..2d506dfd6 100644 --- a/docs/project-memory/shared-memory/pitfalls.md +++ b/docs/project-memory/shared-memory/pitfalls.md @@ -3240,3 +3240,9 @@ - 原因：现役模块的静态 import 会让 Vite、TypeScript 和打包器递归解析依赖；watch ignore 只停止监听，Tailwind source 只控制 class 扫描，tree-shaking 也发生在模块已经加载之后。经 barrel 只取一个公共函数尤其容易把同文件的旧导出一起带回图中。 - 处理：把仍在用的公共账号 / 钱包 / 设置能力迁到明确的现役 client 与 presentation model；Vite `pre` transform 对退役模块真实路径直接失败，ESLint 在现役源上增加 restricted imports。每次恢复公共 UI 后用 `tsc --listFilesOnly` 和全新浏览器 context 复核，不能用已有 HMR 会话判绿。 - 关联：`vite.config.ts`、`.eslintrc.cjs`、`src/services/platform-entry/`、`docs/technical/【架构下线】旧创作模板业务退役方案-2026-07-17.md`。 + +## VectorEngine 请求超时不能脱离 worker 绝对预算（2026-07-20） + +- 现象：VectorEngine 单次请求超时大于 worker job 执行预算时，worker 已停止续租，provider 才超时或开始重试；最终 lease 过期、任务失败并退款，上游却可能继续消耗资源或迟到成功。 +- 原因：单 attempt timeout、重试退避、图片下载与 worker / lease 分别使用独立的相对计时，没有共享同一绝对 deadline；只抬高 worker timeout 或单独压低 provider timeout 都无法保证留出终态写回窗口。 +- 处理：实际调用 VectorEngine 的四类图片 job 使用 `1800s` long 预算；从 job 开始的同一起点派生 provider deadline，常规提前 `60s`、短预算提前一半。每次 attempt、退避、下一次 attempt 和图片下载都必须在该 deadline 内；普通 HTTP / `inline` 不伪造 worker deadline。修复时不改动 lease fencing、迟到写回仲裁和原子退款语义。 diff --git a/docs/technical/【后端架构】外部生成Worker化方案-2026-06-03.md b/docs/technical/【后端架构】外部生成Worker化方案-2026-06-03.md index e546052af..4eb3b79dc 100644 --- a/docs/technical/【后端架构】外部生成Worker化方案-2026-06-03.md +++ b/docs/technical/【后端架构】外部生成Worker化方案-2026-06-03.md @@ -124,7 +124,11 @@ worker 配置： - `GENARRATIVE_EXTERNAL_GENERATION_WORKER_POLL_INTERVAL_MS`：空队列轮询间隔。 - `GENARRATIVE_EXTERNAL_GENERATION_WORKER_LEASE_SECONDS`：任务 lease 时长，默认 `600`；worker 会按约三分之一 lease、最长 30 秒的间隔续租。该值应覆盖一次心跳网络抖动窗口，不需要大于完整外部生成链路耗时。 - `GENARRATIVE_EXTERNAL_GENERATION_WORKER_JOB_TIMEOUT_SECONDS`：普通外部生成 job 的执行预算，默认 `900`。超过预算后当前 worker 停止续租并释放 worker 槽位，但不取消已启动的业务 future，也不主动写入失败 / 重试状态；在途执行交由 lease fencing 仲裁：写回在租约有效期内到达则照常完成，否则被拒绝，租约过期后任务可被重新认领，attempt 耗尽时由认领事务原子标记失败并结算退款。 -- `GENARRATIVE_EXTERNAL_GENERATION_WORKER_LONG_JOB_TIMEOUT_SECONDS`：视频、角色动作等长耗时 job 的执行预算，默认 `1800`。 +- `GENARRATIVE_EXTERNAL_GENERATION_WORKER_LONG_JOB_TIMEOUT_SECONDS`：VectorEngine 图片生成 / 编辑、图标 spritesheet 生成、UI 素材提取以及角色动作、视频等长耗时 job 的执行预算，默认 `1800`。其中四类 VectorEngine 图片 job 固定为 `editor_image_generation`、`editor_image_edit`、`editor_icon_spritesheet_generation` 和 `editor_ui_design_asset_extraction`；手动去背景等不直接调用 VectorEngine 的 job 继续使用普通预算。 + +worker 在单次 job 开始执行时从同一个单调时钟起点计算绝对 `job deadline` 和更早的 `provider deadline`：常规情况下为终态审计、OSS 持久化及 `complete/fail` 回写保留 `60` 秒；当整个 job 预算小于 `120` 秒时，保留其一半，避免 provider 预算被全部吃掉。该 deadline 只通过进程内 `RequestContext` 传给 VectorEngine 图片调用，不写入 HTTP DTO、队列 payload 或 SpacetimeDB；普通 HTTP / `inline` 上下文没有 deadline，保持原有行为。 + +VectorEngine 每次发送的实际 timeout 取 `min(VECTOR_ENGINE_IMAGE_REQUEST_TIMEOUT_MS, provider 剩余预算)`。遇到可重试传输错误或 408 / 429 / 5xx 时，只有当剩余预算还容得下本次退避和下一次 attempt 才继续；否则立即停止重试并返回当前 provider 错误，deadline 耗尽时返回 timeout。同一绝对 deadline 同时覆盖参考图下载、provider 请求 / 响应和响应图片 URL 下载，不允许请求已返回后的图片下载越过 provider 预算。`VECTOR_ENGINE_IMAGE_REQUEST_TIMEOUT_MS` 默认仍为 `1000000`；配置加载层允许显式值低于默认值，不再在读取环境变量时强制抬升。 controller 配置： @@ -136,7 +140,7 @@ controller 配置： - `GENARRATIVE_EXTERNAL_GENERATION_CONTROLLER_SERVICE_TEMPLATE`：systemd worker 模板，默认 `genarrative-external-generation-worker@{}.service`。 - `GENARRATIVE_EXTERNAL_GENERATION_CONTROLLER_DRY_RUN`：只记录决策不执行 systemctl，默认 `false`。 -动态缩扩容方式：生产默认由 `deploy/systemd/genarrative-external-generation-controller.service` 启动 `GENARRATIVE_PROCESS_ROLE=external-generation-controller`，controller 读取 `get_external_generation_queue_stats_and_return` 后对 `genarrative-external-generation-worker@N.service` 执行精确 `systemctl start/stop`；无需改变 HTTP 进程数。controller 只操作 `@1..@MAX` 中的缺口或最高编号多余实例，保留 `@1` 作为保底 worker。缩容或发布重启 worker 时，进程收到 SIGINT/SIGTERM 后会停止 claim 新任务并等待当前任务完成；若进程被硬杀、机器断电或超过 systemd `TimeoutStopSec`，未完成任务会在 lease 过期后被其它 worker 重新领取。若 worker 内业务 future 长时间无返回，执行预算到期后 worker 会停止续租并释放槽位，在途 future 继续运行至租约仲裁窗口；有效租约内的写回仍可完成，租约过期后才会由其它 worker 重新认领，避免客户端取消与服务端写回竞态。容器链路已有独立 `external-generation-worker` compose service；扩 worker 必须扩这个 worker service，不能只扩 `api-server` HTTP service。 +动态缩扩容方式：生产默认由 `deploy/systemd/genarrative-external-generation-controller.service` 启动 `GENARRATIVE_PROCESS_ROLE=external-generation-controller`，controller 读取 `get_external_generation_queue_stats_and_return` 后对 `genarrative-external-generation-worker@N.service` 执行精确 `systemctl start/stop`；无需改变 HTTP 进程数。controller 只操作 `@1..@MAX` 中的缺口或最高编号多余实例，保留 `@1` 作为保底 worker。缩容或发布重启 worker 时，进程收到 SIGINT/SIGTERM 后会停止 claim 新任务并等待当前任务完成；若进程被硬杀、机器断电或超过 systemd `TimeoutStopSec`，未完成任务会在 lease 过期后被其它 worker 重新领取。VectorEngine 图片链路会先于整个 job 执行预算停止 provider 发送 / 重试，以便 worker 在有效 lease 内完成终态写回；若其他业务 future 仍长时间无返回，执行预算到期后 worker 会停止续租并释放槽位，在途 future 继续运行至租约仲裁窗口；有效租约内的写回仍可完成，租约过期后才会由其它 worker 重新认领，避免客户端取消与服务端写回竞态。本次预算收口不改变 lease 续租 / fencing、迟到写回仲裁、attempt 耗尽收口和原子退款语义。容器链路已有独立 `external-generation-worker` compose service；扩 worker 必须扩这个 worker service，不能只扩 `api-server` HTTP service。 ## 已接入的拼图纵切 diff --git a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md index fd027403b..4431efc95 100644 --- a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md +++ b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md @@ -63,7 +63,7 @@ HTTP 角色的 `GENARRATIVE_SPACETIME_POOL_SIZE` 只表示 procedure / reducer 生产拆分角色时，`external-generation-worker` 和 `external-generation-controller` 的专属 env 示例会把 `GENARRATIVE_SPACETIME_POOL_SIZE` 覆盖为 `1`；非 HTTP 角色不创建 API 缓存读连接，只保留 `external_generation_job` 队列窄订阅作为响应式唤醒信号，实际抢占和扩缩容判断仍走 SpacetimeDB procedure。worker / controller 不执行模型定价 seed，启动时先调用受 runtime writer 鉴权的 queue-stats procedure 做只读预检，身份不匹配时 fail-fast；当前正式 systemd unit 通过共同加载 `/etc/genarrative/api-server.env` 继承同一 `GENARRATIVE_SPACETIME_TOKEN`，专属角色 env 示例不重复配置该 token。`GENARRATIVE_EXTERNAL_GENERATION_WORKER_POLL_INTERVAL_MS` 与 controller poll interval 只作为订阅失效、漏事件和 lease 过期这类时间条件的兜底，不作为正常领取任务的主路径。 -生产 worker 默认 `GENARRATIVE_EXTERNAL_GENERATION_WORKER_LEASE_SECONDS=600`，只覆盖 worker 心跳抖动和短暂断连窗口，不再把 lease 当成完整任务时长；默认 `GENARRATIVE_EXTERNAL_GENERATION_WORKER_JOB_TIMEOUT_SECONDS=900`，角色动画 / 视频类长任务使用 `GENARRATIVE_EXTERNAL_GENERATION_WORKER_LONG_JOB_TIMEOUT_SECONDS=1800`。worker 在单次尝试超过执行预算后会停止续租并释放 worker 槽位，但不会取消已启动的业务 future 或主动写入失败 / 重试状态；在途执行由 lease fencing 仲裁，有效租约内写回仍可完成，租约过期后任务才可重新领取，attempt 耗尽时由认领事务标记失败并结算退款。生产部署和 provision 脚本会给 `/etc/genarrative/api-server.env` 与 `/etc/genarrative/external-generation-worker.env` 补齐这些变量；已有自定义值不覆盖，只会把历史旧默认 `3600` 迁移为 `600`。 +生产 worker 默认 `GENARRATIVE_EXTERNAL_GENERATION_WORKER_LEASE_SECONDS=600`，只覆盖 worker 心跳抖动和短暂断连窗口，不再把 lease 当成完整任务时长；默认 `GENARRATIVE_EXTERNAL_GENERATION_WORKER_JOB_TIMEOUT_SECONDS=900`。`editor_image_generation`、`editor_image_edit`、`editor_icon_spritesheet_generation`、`editor_ui_design_asset_extraction` 四类 VectorEngine 图片任务与角色动画 / 视频类长任务使用 `GENARRATIVE_EXTERNAL_GENERATION_WORKER_LONG_JOB_TIMEOUT_SECONDS=1800`，手动去背景、音效和背景音乐继续使用普通预算。worker 在单次尝试超过执行预算后会停止续租并释放 worker 槽位，但不会取消已启动的业务 future 或主动写入失败 / 重试状态；在途执行由 lease fencing 仲裁，有效租约内写回仍可完成，租约过期后任务才可重新领取，attempt 耗尽时由认领事务标记失败并结算退款。生产部署和 provision 脚本会给 `/etc/genarrative/api-server.env` 与 `/etc/genarrative/external-generation-worker.env` 补齐这些变量；已有自定义值不覆盖，只会把历史旧默认 `3600` 迁移为 `600`。 lease 过期后不代表任务一定再次执行：claim transaction 只有在 `attempt < max_attempts` 时才会递增 attempt 并返回 worker；如果过期的是最终 attempt，则直接把 job 收口为 `failed`、清理 lease，并按入队冻结价格为当前 attempt 原子退款或写 cancellation intent。该终态任务不会再次进入 provider executor，迟到 consume 会被 settlement intent 拒绝。 @@ -153,9 +153,9 @@ spacetime sql "SELECT * FROM runtime_setting LIMIT 1" --server http:/ 本地 `spacetime` CLI / standalone 版本必须和 `server-rs/Cargo.toml` 里锁定的 `spacetimedb` 版本一致；当前统一版本为 `2.6.1`。若版本错配，procedure 返回值可能在宿主侧触发 `Failed to BSATN deserialize procedure return value`，api-server 最终表现为现役 settings、editor project 或 profile procedure 超时。排障时先运行 `spacetime --version`，再对照 `server-rs/Cargo.toml` 的 `spacetimedb = "..."`；遇到版本不匹配时直接执行 `spacetime version install && spacetime version use `，或在目标就是最新版本时执行 `spacetime version upgrade`，升级后重启 `npm run dev:spacetime` 再重试。当前 `scripts/dev.mjs` 会在启动和复用本地 SpacetimeDB 前写入并校验 `dev-spacetime-tool-version`。2.6.1 修复了 procedure context 中调用者 `Identity` / `ConnectionId` 始终为空的回归，依赖 `ctx.sender` 鉴权时必须同时确认宿主已升级。 -本地 `.env`、`.env.local` 或 `.env.secrets.local` 修改后必须重启 `api-server` 才会生效；若已经通过 `npm run dev` 启动完整联调，可在该终端输入 `rs api-server`。排查图片编辑器 VectorEngine 生成链路时，确认 `VECTOR_ENGINE_BASE_URL`、`VECTOR_ENGINE_API_KEY` 和 `VECTOR_ENGINE_IMAGE_REQUEST_TIMEOUT_MS` 只在本地或服务器密钥文件中配置，不能写入 Git。VectorEngine `gpt-image-2` 图片协议、URL / base64 响应解析、远端图片下载和 provider 侧结构化日志在 `server-rs/crates/platform-image`；`api-server` 只做编辑器请求编排、OSS / asset 持久化、计费和失败审计落库。`platform-image` 会在 JSON 生成和 multipart 编辑请求发送前归一显式像素尺寸；若请求发送失败，先按同一 `request_id` 查看 provider 日志与 `external_api_call_failure.metadata_json.errorSource`，当前 multipart `/v1/images/edits` 单独强制 HTTP/1.1。 +本地 `.env`、`.env.local` 或 `.env.secrets.local` 修改后必须重启 `api-server` 才会生效；若已经通过 `npm run dev` 启动完整联调，可在该终端输入 `rs api-server`。排查图片编辑器 VectorEngine 生成链路时，确认 `VECTOR_ENGINE_BASE_URL`、`VECTOR_ENGINE_API_KEY` 和 `VECTOR_ENGINE_IMAGE_REQUEST_TIMEOUT_MS` 只在本地或服务器密钥文件中配置，不能写入 Git。`VECTOR_ENGINE_IMAGE_REQUEST_TIMEOUT_MS` 是单次 attempt 的配置上限，默认 `1000000`；配置加载层允许显式值低于该默认值，不再在读取环境变量时强制抬高。VectorEngine `gpt-image-2` 图片协议、URL / base64 响应解析、远端图片下载和 provider 侧结构化日志在 `server-rs/crates/platform-image`；`api-server` 只做编辑器请求编排、OSS / asset 持久化、计费和失败审计落库。`platform-image` 会在 JSON 生成和 multipart 编辑请求发送前归一显式像素尺寸；若请求发送失败，先按同一 `request_id` 查看 provider 日志与 `external_api_call_failure.metadata_json.errorSource`，当前 multipart `/v1/images/edits` 单独强制 HTTP/1.1。 -VectorEngine 图片生成 / 编辑在 `request_send` 阶段出现 `timeout`、`connect`、libcurl 35 SSL connect reset、libcurl 56 receive error / `unexpected eof while reading`、recv failure 等临时传输错误，或在 `upstream_status` 阶段收到 408 / 429 / 5xx（例如 Nginx HTML `502 Bad Gateway`）时，`platform-image` 会对同一请求最多发送 5 次；multipart 图片编辑每次重试都会重新构造 form，避免复用已消费的 body。日志中 `VectorEngine 图片请求发送失败，准备重试` 或 `VectorEngine 图片上游状态可重试，准备重试` 表示本次失败已进入下一次尝试；最终仍失败时才会写入 `external_api_call_failure` 并返回 504 / 502。排查生产失败时应同时统计 retry 前的尝试日志和最终 audit，避免把一次用户请求内的多次发送误判成多个用户请求。 +VectorEngine 图片生成 / 编辑在 `request_send` 阶段出现 `timeout`、`connect`、libcurl 35 SSL connect reset、libcurl 56 receive error / `unexpected eof while reading`、recv failure 等临时传输错误，或在 `upstream_status` 阶段收到 408 / 429 / 5xx（例如 Nginx HTML `502 Bad Gateway`）时，`platform-image` 会对同一请求最多发送 5 次；multipart 图片编辑每次重试都会重新构造 form，避免复用已消费的 body。worker 从 job 开始的同一时钟起点计算绝对 deadline，常规保留最后 `60` 秒给审计、OSS 和终态写回；job 预算小于 `120` 秒时保留一半。VectorEngine 单次 attempt timeout 取配置值和剩余 provider 预算的较小值；退避后已没有下一次 attempt 的预算时立即停止重试。该 deadline 覆盖参考图、provider 请求 / 响应和响应图片下载的整次 provider future，但只在 worker 进程内通过 `RequestContext` 传递；普通 HTTP / `inline` 没有该 deadline，继续保持原有 timeout 和重试行为。日志中 `VectorEngine 图片请求发送失败，准备重试` 或 `VectorEngine 图片上游状态可重试，准备重试` 表示本次失败确有预算进入下一次尝试；预算耗尽或最终仍失败时才会写入 `external_api_call_failure` 并返回 504 / 502。排查生产失败时应同时统计 retry 前的尝试日志和最终 audit，避免把一次用户请求内的多次发送误判成多个用户请求。这项收口不修改 lease 续租 / fencing、迟到写回仲裁、attempt 耗尽与原子退款语义。 图片编辑器生成属于持久队列长任务：提交接口返回 job 后，前端通过 `/api/runtime/external-generation/jobs/{jobId}` 与编辑器项目资源状态收敛。生产排查小程序或 WebView `Failed to fetch` 时，若 Nginx access log 为 `499`、`upstream_status=-`，先按提交请求的 `request_id`、job id、worker 日志和 `external_api_call_failure` 对齐真实任务，不把客户端断开直接判定为 provider 失败。 diff --git a/server-rs/crates/api-server/src/config.rs b/server-rs/crates/api-server/src/config.rs index 397366ee8..e4cb6de96 100644 --- a/server-rs/crates/api-server/src/config.rs +++ b/server-rs/crates/api-server/src/config.rs @@ -1090,9 +1090,8 @@ impl AppConfig { if let Some(vector_engine_image_request_timeout_ms) = read_first_positive_u64_env(&["VECTOR_ENGINE_IMAGE_REQUEST_TIMEOUT_MS"]) { - // 中文注释：VectorEngine image-2 实测可能超过 500 秒；旧环境文件中常见的 180 秒值不能再提前截断真实生图。 - config.vector_engine_image_request_timeout_ms = vector_engine_image_request_timeout_ms - .max(DEFAULT_VECTOR_ENGINE_IMAGE_REQUEST_TIMEOUT_MS); + // 单次 attempt 上限允许按环境收短；worker 调用还会受整次任务的绝对 deadline 约束。 + config.vector_engine_image_request_timeout_ms = vector_engine_image_request_timeout_ms; } if let Some(vector_engine_audio_request_timeout_ms) = @@ -1530,9 +1529,8 @@ mod tests { DEFAULT_EDITOR_BGFILTER_REQUEST_TIMEOUT_MS, DEFAULT_EXTERNAL_GENERATION_WORKER_JOB_TIMEOUT_SECONDS, DEFAULT_EXTERNAL_GENERATION_WORKER_LEASE_SECONDS, - DEFAULT_EXTERNAL_GENERATION_WORKER_LONG_JOB_TIMEOUT_SECONDS, - DEFAULT_VECTOR_ENGINE_IMAGE_REQUEST_TIMEOUT_MS, ExternalGenerationMode, LlmProvider, - ProcessRole, parse_bool, parse_external_generation_mode, parse_process_role, + DEFAULT_EXTERNAL_GENERATION_WORKER_LONG_JOB_TIMEOUT_SECONDS, ExternalGenerationMode, + LlmProvider, ProcessRole, parse_bool, parse_external_generation_mode, parse_process_role, }; use std::{ fs, @@ -1773,10 +1771,7 @@ mod tests { config.vector_engine_base_url, "https://vector.internal.example" ); - assert_eq!( - config.vector_engine_image_request_timeout_ms, - DEFAULT_VECTOR_ENGINE_IMAGE_REQUEST_TIMEOUT_MS - ); + assert_eq!(config.vector_engine_image_request_timeout_ms, 210_000); assert_eq!( config.hyper3d_base_url, "https://model.internal.example/api/v2" diff --git a/server-rs/crates/api-server/src/edutainment_baby_object.rs b/server-rs/crates/api-server/src/edutainment_baby_object.rs index abfb53493..996a9073a 100644 --- a/server-rs/crates/api-server/src/edutainment_baby_object.rs +++ b/server-rs/crates/api-server/src/edutainment_baby_object.rs @@ -1049,6 +1049,7 @@ mod tests { base_url: "https://vector.example".to_string(), api_key: "secret".to_string(), request_timeout_ms: 180_000, + request_deadline: None, external_api_audit_state: None, external_api_audit_user_id: None, external_api_audit_profile_id: None, diff --git a/server-rs/crates/api-server/src/external_generation_worker.rs b/server-rs/crates/api-server/src/external_generation_worker.rs index 8d07a2004..ea682e4a1 100644 --- a/server-rs/crates/api-server/src/external_generation_worker.rs +++ b/server-rs/crates/api-server/src/external_generation_worker.rs @@ -1,4 +1,9 @@ -use std::{future::Future, io, pin::Pin, time::Duration}; +use std::{ + future::Future, + io, + pin::Pin, + time::{Duration, Instant}, +}; use axum::Json; use serde_json::{Value, json}; @@ -15,6 +20,8 @@ use tokio::{ use tracing::{error, info, warn}; const MAX_EDITOR_GENERATION_WARNING_CHARS: usize = 2_048; +// provider 必须先结束，给失败审计、计费结算和队列终态写回保留有效 lease 内的收尾窗口。 +const EXTERNAL_GENERATION_WORKER_TERMINAL_WRITE_RESERVE: Duration = Duration::from_secs(60); const EDITOR_GENERATION_SLICE_WARNING_PREFIX: &str = "图集已生成，但自动拆分未完成："; const EDITOR_GENERATION_WARNING_REDACTED_MESSAGE: &str = "自动拆分未完成（告警详情含内联媒体引用，已省略）"; @@ -315,6 +322,8 @@ async fn process_external_generation_job( ) -> Result<(), String> { let heartbeat_interval = external_generation_worker_heartbeat_interval(lease); let job_timeout = external_generation_worker_job_timeout(&state.config, job.job_kind.as_str()); + let (job_deadline, provider_deadline) = + external_generation_worker_deadlines(Instant::now(), job_timeout); let billing_price_mud_points = match external_generation_billing_price_mud_points(&job) { Ok(price_mud_points) => price_mud_points, Err(message) => { @@ -338,14 +347,19 @@ async fn process_external_generation_job( job.job_id.clone(), billing_claim_attempt, billing_price_mud_points, - process_external_generation_job_once(state.clone(), worker_id.clone(), job.clone()), + process_external_generation_job_once( + state.clone(), + worker_id.clone(), + job.clone(), + provider_deadline, + ), )); let heartbeat = maintain_external_generation_job_lease(&state, &worker_id, &job, lease, heartbeat_interval); match await_external_generation_job_execution( join_external_generation_work(&mut work_handle), heartbeat, - job_timeout, + job_deadline, ) .await { @@ -446,7 +460,7 @@ enum ExternalGenerationJobExecutionOutcome { async fn await_external_generation_job_execution( work: W, heartbeat: H, - job_timeout: Duration, + job_deadline: Instant, ) -> ExternalGenerationJobExecutionOutcome where W: Future>, @@ -454,13 +468,13 @@ where { tokio::pin!(work); tokio::pin!(heartbeat); - let job_deadline = sleep(job_timeout); - tokio::pin!(job_deadline); + let job_deadline_sleep = tokio::time::sleep_until(job_deadline.into()); + tokio::pin!(job_deadline_sleep); tokio::select! { biased; result = &mut work => ExternalGenerationJobExecutionOutcome::Finished(result), - _ = &mut job_deadline => ExternalGenerationJobExecutionOutcome::TimedOut, + _ = &mut job_deadline_sleep => ExternalGenerationJobExecutionOutcome::TimedOut, result = &mut heartbeat => match result { Ok(()) => unreachable!("external generation heartbeat monitor should not finish"), Err(error) => ExternalGenerationJobExecutionOutcome::LeaseRenewalFailed(error), @@ -485,6 +499,7 @@ async fn process_external_generation_job_once( state: AppState, worker_id: String, job: ExternalGenerationJobRecord, + provider_deadline: Instant, ) -> Result<(), String> { match job.job_kind.as_str() { #[cfg(any())] @@ -752,7 +767,7 @@ async fn process_external_generation_job_once( return Err(message); } }; - let request_context = worker_request_context(&job); + let request_context = worker_request_context(&job, provider_deadline); match generate_editor_image_for_owner( &state, &request_context, @@ -788,7 +803,7 @@ async fn process_external_generation_job_once( return Err(message); } }; - let request_context = worker_request_context(&job); + let request_context = worker_request_context(&job, provider_deadline); match edit_editor_image_for_owner( &state, &request_context, @@ -819,7 +834,7 @@ async fn process_external_generation_job_once( } }; payload.task_id.get_or_insert_with(|| job.job_id.clone()); - let request_context = worker_request_context(&job); + let request_context = worker_request_context(&job, provider_deadline); match remove_editor_image_background_for_owner( &state, &request_context, @@ -849,7 +864,7 @@ async fn process_external_generation_job_once( return Err(message); } }; - let request_context = worker_request_context(&job); + let request_context = worker_request_context(&job, provider_deadline); match generate_editor_icon_spritesheet_for_owner( &state, &request_context, @@ -885,7 +900,7 @@ async fn process_external_generation_job_once( return Err(message); } }; - let request_context = worker_request_context(&job); + let request_context = worker_request_context(&job, provider_deadline); match extract_editor_ui_design_assets_for_owner( &state, &request_context, @@ -922,7 +937,7 @@ async fn process_external_generation_job_once( return Err(message); } }; - let request_context = worker_request_context(&job); + let request_context = worker_request_context(&job, provider_deadline); match generate_editor_character_animation_for_owner( state.clone(), request_context, @@ -954,7 +969,7 @@ async fn process_external_generation_job_once( return Err(message); } }; - let request_context = worker_request_context(&job); + let request_context = worker_request_context(&job, provider_deadline); match generate_editor_video_for_owner( state.clone(), request_context, @@ -985,7 +1000,7 @@ async fn process_external_generation_job_once( return Err(message); } }; - let request_context = worker_request_context(&job); + let request_context = worker_request_context(&job, provider_deadline); match generate_editor_sound_effect_for_owner( state.clone(), request_context, @@ -1016,7 +1031,7 @@ async fn process_external_generation_job_once( return Err(message); } }; - let request_context = worker_request_context(&job); + let request_context = worker_request_context(&job, provider_deadline); match generate_editor_background_music_for_owner( state.clone(), request_context, @@ -1112,13 +1127,17 @@ async fn fail_queue_job_after_worker_error( Ok(()) } -fn worker_request_context(job: &ExternalGenerationJobRecord) -> RequestContext { +fn worker_request_context( + job: &ExternalGenerationJobRecord, + provider_deadline: Instant, +) -> RequestContext { RequestContext::new( format!("external-generation-worker-{}", job.job_id), format!("external-generation-worker {}", job.job_kind), std::time::Duration::ZERO, false, ) + .with_external_call_deadline(provider_deadline) } fn editor_generation_worker_caller( @@ -1409,11 +1428,29 @@ fn external_generation_worker_heartbeat_interval(lease: Duration) -> Duration { Duration::from_millis(heartbeat_millis) } +fn external_generation_worker_deadlines( + started_at: Instant, + job_timeout: Duration, +) -> (Instant, Instant) { + let terminal_write_reserve = + EXTERNAL_GENERATION_WORKER_TERMINAL_WRITE_RESERVE.min(job_timeout / 2); + let Some(job_deadline) = started_at.checked_add(job_timeout) else { + return (started_at, started_at); + }; + let provider_deadline = started_at + .checked_add(job_timeout.saturating_sub(terminal_write_reserve)) + .unwrap_or(started_at); + (job_deadline, provider_deadline) +} + fn external_generation_worker_job_timeout(config: &AppConfig, job_kind: &str) -> Duration { match job_kind { - EDITOR_CHARACTER_ANIMATION_GENERATION_JOB_KIND | EDITOR_VIDEO_GENERATION_JOB_KIND => { - config.external_generation_worker_long_job_timeout - } + EDITOR_IMAGE_GENERATION_JOB_KIND + | EDITOR_IMAGE_EDIT_JOB_KIND + | EDITOR_ICON_SPRITESHEET_GENERATION_JOB_KIND + | EDITOR_UI_DESIGN_ASSET_EXTRACTION_JOB_KIND + | EDITOR_CHARACTER_ANIMATION_GENERATION_JOB_KIND + | EDITOR_VIDEO_GENERATION_JOB_KIND => config.external_generation_worker_long_job_timeout, _ => config.external_generation_worker_job_timeout, } } @@ -1556,8 +1593,12 @@ mod tests { std::future::pending::>().await }; - let outcome = - await_external_generation_job_execution(work, heartbeat, Duration::from_secs(1)).await; + let outcome = await_external_generation_job_execution( + work, + heartbeat, + Instant::now() + Duration::from_secs(1), + ) + .await; assert_eq!( outcome, @@ -1578,9 +1619,12 @@ mod tests { }; let heartbeat = std::future::pending::>(); - let outcome = - await_external_generation_job_execution(work, heartbeat, Duration::from_millis(25)) - .await; + let outcome = await_external_generation_job_execution( + work, + heartbeat, + Instant::now() + Duration::from_millis(25), + ) + .await; assert_eq!(outcome, ExternalGenerationJobExecutionOutcome::TimedOut); assert!( @@ -1884,7 +1928,7 @@ mod tests { } #[test] - fn worker_job_timeout_uses_long_budget_for_video_jobs() { + fn worker_job_timeout_uses_long_budget_for_image_and_video_jobs() { let config = AppConfig { external_generation_worker_job_timeout: Duration::from_secs(60), external_generation_worker_long_job_timeout: Duration::from_secs(600), @@ -1893,7 +1937,25 @@ mod tests { assert_eq!( external_generation_worker_job_timeout(&config, EDITOR_IMAGE_GENERATION_JOB_KIND), - Duration::from_secs(60) + Duration::from_secs(600) + ); + assert_eq!( + external_generation_worker_job_timeout(&config, EDITOR_IMAGE_EDIT_JOB_KIND), + Duration::from_secs(600) + ); + assert_eq!( + external_generation_worker_job_timeout( + &config, + EDITOR_ICON_SPRITESHEET_GENERATION_JOB_KIND + ), + Duration::from_secs(600) + ); + assert_eq!( + external_generation_worker_job_timeout( + &config, + EDITOR_UI_DESIGN_ASSET_EXTRACTION_JOB_KIND + ), + Duration::from_secs(600) ); assert_eq!( external_generation_worker_job_timeout( @@ -1906,6 +1968,61 @@ mod tests { external_generation_worker_job_timeout(&config, EDITOR_VIDEO_GENERATION_JOB_KIND), Duration::from_secs(600) ); + assert_eq!( + external_generation_worker_job_timeout(&config, EDITOR_BACKGROUND_REMOVAL_JOB_KIND), + Duration::from_secs(60) + ); + assert_eq!( + external_generation_worker_job_timeout( + &config, + EDITOR_SOUND_EFFECT_GENERATION_JOB_KIND + ), + Duration::from_secs(60) + ); + } + + #[test] + fn worker_provider_deadline_reserves_terminal_write_budget() { + let started_at = Instant::now(); + let (job_deadline, provider_deadline) = + external_generation_worker_deadlines(started_at, Duration::from_secs(1_800)); + + assert_eq!( + job_deadline.duration_since(started_at), + Duration::from_secs(1_800) + ); + assert_eq!( + provider_deadline.duration_since(started_at), + Duration::from_secs(1_740) + ); + assert_eq!( + job_deadline.duration_since(provider_deadline), + Duration::from_secs(60) + ); + + let (short_job_deadline, short_provider_deadline) = + external_generation_worker_deadlines(started_at, Duration::from_secs(30)); + assert_eq!( + short_job_deadline.duration_since(started_at), + Duration::from_secs(30) + ); + assert_eq!( + short_provider_deadline.duration_since(started_at), + Duration::from_secs(15) + ); + } + + #[test] + fn worker_request_context_carries_provider_deadline() { + let job = external_generation_job_record_fixture(Some("lease-1")); + let provider_deadline = Instant::now() + Duration::from_secs(30); + + let request_context = worker_request_context(&job, provider_deadline); + + assert_eq!( + request_context.external_call_deadline(), + Some(provider_deadline) + ); } #[test] diff --git a/server-rs/crates/api-server/src/jump_hop.rs b/server-rs/crates/api-server/src/jump_hop.rs index a689909d0..254871434 100644 --- a/server-rs/crates/api-server/src/jump_hop.rs +++ b/server-rs/crates/api-server/src/jump_hop.rs @@ -3005,6 +3005,7 @@ mod tests { base_url: base_url.clone(), api_key: api_key.clone(), request_timeout_ms: 180_000, + request_deadline: None, }; let http_client = platform_image::build_vector_engine_image_http_client(&settings) .expect("构建 HTTP 客户端"); @@ -3324,6 +3325,7 @@ mod tests { base_url: base_url.clone(), api_key: api_key.clone(), request_timeout_ms: 180_000, + request_deadline: None, }; let http_client = platform_image::build_vector_engine_image_http_client(&settings) .expect("构建 HTTP 客户端"); diff --git a/server-rs/crates/api-server/src/openai_image_generation.rs b/server-rs/crates/api-server/src/openai_image_generation.rs index 6e461dfd0..55ee30c46 100644 --- a/server-rs/crates/api-server/src/openai_image_generation.rs +++ b/server-rs/crates/api-server/src/openai_image_generation.rs @@ -13,6 +13,7 @@ use platform_image::{ vector_engine_images_generation_url, }; use serde_json::{Value, json}; +use std::time::Instant; use time::OffsetDateTime; use crate::{ @@ -39,6 +40,7 @@ pub(crate) struct OpenAiImageSettings { pub base_url: String, pub api_key: String, pub request_timeout_ms: u64, + pub request_deadline: Option, pub external_api_audit_state: Option, pub external_api_audit_user_id: Option, pub external_api_audit_profile_id: Option, @@ -52,6 +54,7 @@ impl std::fmt::Debug for OpenAiImageSettings { .field("base_url", &self.base_url) .field("api_key", &"") .field("request_timeout_ms", &self.request_timeout_ms) + .field("request_deadline_enabled", &self.request_deadline.is_some()) .field( "external_api_audit_enabled", &self.external_api_audit_state.is_some(), @@ -107,6 +110,7 @@ pub(crate) fn require_openai_image_settings( base_url: base_url.to_string(), api_key: api_key.to_string(), request_timeout_ms: state.config.vector_engine_image_request_timeout_ms.max(1), + request_deadline: None, external_api_audit_state: Some(state.clone()), external_api_audit_user_id: None, external_api_audit_profile_id: None, @@ -406,6 +410,7 @@ impl OpenAiImageSettings { self.external_api_audit_user_id = user_id; self.external_api_audit_profile_id = profile_id; self.external_api_audit_request_id = Some(request_context.request_id().to_string()); + self.request_deadline = request_context.external_call_deadline(); self } @@ -414,6 +419,7 @@ impl OpenAiImageSettings { base_url: self.base_url.clone(), api_key: self.api_key.clone(), request_timeout_ms: self.request_timeout_ms.max(1), + request_deadline: self.request_deadline, } } } @@ -573,6 +579,35 @@ mod tests { use super::*; use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD}; + #[test] + fn external_api_audit_context_forwards_external_call_deadline() { + let deadline = Instant::now() + std::time::Duration::from_secs(30); + let request_context = RequestContext::new( + "request-1".to_string(), + "external-generation-worker editor_image_edit".to_string(), + std::time::Duration::ZERO, + false, + ) + .with_external_call_deadline(deadline); + let settings = OpenAiImageSettings { + base_url: "https://vector.example".to_string(), + api_key: "test-key".to_string(), + request_timeout_ms: 1_000_000, + request_deadline: None, + external_api_audit_state: None, + external_api_audit_user_id: None, + external_api_audit_profile_id: None, + external_api_audit_request_id: None, + } + .with_external_api_audit_context(&request_context, None, None); + + assert_eq!(settings.request_deadline, Some(deadline)); + assert_eq!( + settings.provider_settings().request_deadline, + Some(deadline) + ); + } + #[test] fn gpt_image_2_generation_request_uses_create_model_without_reference_images() { let body = build_openai_image_request_body( @@ -597,6 +632,7 @@ mod tests { base_url: "https://vector.example".to_string(), api_key: "test-key".to_string(), request_timeout_ms: 1_000_000, + request_deadline: None, external_api_audit_state: None, external_api_audit_user_id: None, external_api_audit_profile_id: None, @@ -606,6 +642,7 @@ mod tests { base_url: "https://vector.example/v1".to_string(), api_key: "test-key".to_string(), request_timeout_ms: 1_000_000, + request_deadline: None, external_api_audit_state: None, external_api_audit_user_id: None, external_api_audit_profile_id: None, @@ -628,6 +665,7 @@ mod tests { base_url: "https://vector.example".to_string(), api_key: "test-key".to_string(), request_timeout_ms: 1_000_000, + request_deadline: None, external_api_audit_state: None, external_api_audit_user_id: None, external_api_audit_profile_id: None, @@ -637,6 +675,7 @@ mod tests { base_url: "https://vector.example/v1".to_string(), api_key: "test-key".to_string(), request_timeout_ms: 1_000_000, + request_deadline: None, external_api_audit_state: None, external_api_audit_user_id: None, external_api_audit_profile_id: None, @@ -659,6 +698,7 @@ mod tests { base_url: "https://vector.example".to_string(), api_key: "test-key".to_string(), request_timeout_ms: 1_000_000, + request_deadline: None, external_api_audit_state: None, external_api_audit_user_id: None, external_api_audit_profile_id: None, diff --git a/server-rs/crates/api-server/src/puzzle/vector_engine.rs b/server-rs/crates/api-server/src/puzzle/vector_engine.rs index 285e31054..11ac1469a 100644 --- a/server-rs/crates/api-server/src/puzzle/vector_engine.rs +++ b/server-rs/crates/api-server/src/puzzle/vector_engine.rs @@ -30,6 +30,7 @@ pub(crate) struct PuzzleVectorEngineSettings { pub(crate) base_url: String, pub(crate) api_key: String, pub(crate) request_timeout_ms: u64, + pub(crate) request_deadline: Option, pub(crate) external_api_audit_state: Option, pub(crate) external_api_audit_user_id: Option, pub(crate) external_api_audit_profile_id: Option, @@ -101,6 +102,7 @@ impl PuzzleVectorEngineSettings { base_url: self.base_url.clone(), api_key: self.api_key.clone(), request_timeout_ms: self.request_timeout_ms, + request_deadline: self.request_deadline, external_api_audit_state: self.external_api_audit_state.clone(), external_api_audit_user_id: self.external_api_audit_user_id.clone(), external_api_audit_profile_id: self.external_api_audit_profile_id.clone(), @@ -117,6 +119,7 @@ impl PuzzleVectorEngineSettings { self.external_api_audit_user_id = user_id; self.external_api_audit_profile_id = profile_id; self.external_api_audit_request_id = Some(request_context.request_id().to_string()); + self.request_deadline = request_context.external_call_deadline(); self } } @@ -193,6 +196,7 @@ pub(crate) fn require_puzzle_vector_engine_settings( base_url: base_url.to_string(), api_key: api_key.to_string(), request_timeout_ms: state.vector_engine_image_request_timeout_ms().max(1), + request_deadline: None, external_api_audit_state: Some(state.root_state().clone()), external_api_audit_user_id: None, external_api_audit_profile_id: None, diff --git a/server-rs/crates/api-server/src/request_context.rs b/server-rs/crates/api-server/src/request_context.rs index bb94a66f3..ebd597b39 100644 --- a/server-rs/crates/api-server/src/request_context.rs +++ b/server-rs/crates/api-server/src/request_context.rs @@ -18,6 +18,7 @@ pub struct RequestContext { operation: String, request_started_at: Instant, wants_envelope: bool, + external_call_deadline: Option, } impl RequestContext { @@ -34,9 +35,15 @@ impl RequestContext { .checked_sub(elapsed_seed) .unwrap_or_else(Instant::now), wants_envelope, + external_call_deadline: None, } } + pub fn with_external_call_deadline(mut self, deadline: Instant) -> Self { + self.external_call_deadline = Some(deadline); + self + } + pub fn request_id(&self) -> &str { &self.request_id } @@ -49,6 +56,10 @@ impl RequestContext { self.wants_envelope } + pub fn external_call_deadline(&self) -> Option { + self.external_call_deadline + } + pub fn elapsed(&self) -> u64 { self.request_started_at .elapsed() @@ -111,3 +122,20 @@ fn wants_api_envelope(request: &HttpRequest) -> bool { .map(str::to_lowercase) .is_some_and(|value| matches!(value.as_str(), "1" | "true" | "v1" | "envelope")) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn request_context_has_no_external_call_deadline_by_default() { + let context = RequestContext::new( + "request-1".to_string(), + "GET /healthz".to_string(), + Duration::ZERO, + false, + ); + + assert_eq!(context.external_call_deadline(), None); + } +} diff --git a/server-rs/crates/platform-image/src/vector_engine/budget.rs b/server-rs/crates/platform-image/src/vector_engine/budget.rs new file mode 100644 index 000000000..a7c82ec46 --- /dev/null +++ b/server-rs/crates/platform-image/src/vector_engine/budget.rs @@ -0,0 +1,179 @@ +use std::time::{Duration, Instant}; + +use super::{ + audit::build_failure_audit, constants::VECTOR_ENGINE_PROVIDER, error::PlatformImageError, +}; + +pub(crate) fn effective_request_timeout_ms( + configured_timeout_ms: u64, + request_deadline: Option, +) -> Option { + effective_request_timeout_ms_at(configured_timeout_ms, request_deadline, Instant::now()) +} + +fn effective_request_timeout_ms_at( + configured_timeout_ms: u64, + request_deadline: Option, + now: Instant, +) -> Option { + let configured_timeout_ms = configured_timeout_ms.max(1); + let Some(request_deadline) = request_deadline else { + return Some(configured_timeout_ms); + }; + let remaining = request_deadline.checked_duration_since(now)?; + let remaining_ms = u64::try_from(remaining.as_millis()).unwrap_or(u64::MAX); + if remaining_ms == 0 { + return None; + } + Some(configured_timeout_ms.min(remaining_ms)) +} + +pub(crate) fn retry_delay_fits_request_deadline( + request_deadline: Option, + delay_ms: u64, +) -> bool { + retry_delay_fits_request_deadline_at(request_deadline, delay_ms, Instant::now()) +} + +fn retry_delay_fits_request_deadline_at( + request_deadline: Option, + delay_ms: u64, + now: Instant, +) -> bool { + let Some(request_deadline) = request_deadline else { + return true; + }; + let Some(remaining) = request_deadline.checked_duration_since(now) else { + return false; + }; + let required = Duration::from_millis(delay_ms).saturating_add(Duration::from_millis(1)); + remaining >= required +} + +pub(crate) fn request_budget_exhausted_error( + request_url: &str, + operation: &str, + latency_ms: Option, + prompt_chars: Option, + reference_image_count: Option, +) -> PlatformImageError { + const ERROR_SOURCE: &str = "external request deadline elapsed"; + let message = format!("{operation}：外部图片请求执行预算已耗尽"); + let audit = build_failure_audit( + request_url, + operation, + "request_budget", + None, + None, + true, + false, + message.as_str(), + Some(ERROR_SOURCE.to_string()), + None, + latency_ms, + prompt_chars, + reference_image_count, + ); + tracing::warn!( + provider = VECTOR_ENGINE_PROVIDER, + endpoint = %request_url, + failure_stage = "request_budget", + timeout = true, + elapsed_ms = latency_ms, + prompt_chars, + reference_image_count, + operation, + "VectorEngine 图片请求执行预算已耗尽" + ); + PlatformImageError::Request { + provider: VECTOR_ENGINE_PROVIDER, + message, + endpoint: Some(request_url.to_string()), + timeout: true, + connect: false, + request: true, + body: false, + status_code: None, + source: Some(ERROR_SOURCE.to_string()), + audit: Some(audit), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn configured_timeout_is_preserved_without_deadline() { + let now = Instant::now(); + + assert_eq!( + effective_request_timeout_ms_at(1_000, None, now), + Some(1_000) + ); + assert_eq!(effective_request_timeout_ms_at(0, None, now), Some(1)); + } + + #[test] + fn request_timeout_is_clipped_to_remaining_deadline_budget() { + let now = Instant::now(); + + assert_eq!( + effective_request_timeout_ms_at(10_000, Some(now + Duration::from_millis(250)), now,), + Some(250) + ); + assert_eq!( + effective_request_timeout_ms_at(100, Some(now + Duration::from_millis(250)), now,), + Some(100) + ); + } + + #[test] + fn exhausted_or_sub_millisecond_deadline_stops_request_before_send() { + let now = Instant::now(); + + assert_eq!(effective_request_timeout_ms_at(1_000, Some(now), now), None); + assert_eq!( + effective_request_timeout_ms_at(1_000, Some(now + Duration::from_micros(999)), now,), + None + ); + } + + #[test] + fn retry_requires_budget_for_backoff_and_next_attempt() { + let now = Instant::now(); + + assert!(retry_delay_fits_request_deadline_at(None, 500, now)); + assert!(retry_delay_fits_request_deadline_at( + Some(now + Duration::from_millis(501)), + 500, + now, + )); + assert!(!retry_delay_fits_request_deadline_at( + Some(now + Duration::from_millis(500)), + 500, + now, + )); + assert!(!retry_delay_fits_request_deadline_at(Some(now), 0, now)); + } + + #[test] + fn exhausted_budget_maps_to_timeout_request_error_and_audit() { + let error = request_budget_exhausted_error( + "https://vector.example/v1/images/generations", + "生成图片失败", + Some(900), + Some(12), + Some(1), + ); + + assert!(matches!( + error, + PlatformImageError::Request { timeout: true, .. } + )); + let audit = error.audit().expect("budget error should carry audit"); + assert_eq!(audit.failure_stage, "request_budget"); + assert!(audit.timeout); + assert_eq!(audit.latency_ms, Some(900)); + } +} diff --git a/server-rs/crates/platform-image/src/vector_engine/client.rs b/server-rs/crates/platform-image/src/vector_engine/client.rs index d8ee9271d..54647031f 100644 --- a/server-rs/crates/platform-image/src/vector_engine/client.rs +++ b/server-rs/crates/platform-image/src/vector_engine/client.rs @@ -1,10 +1,14 @@ -use std::time::{SystemTime, UNIX_EPOCH}; +use std::time::{Instant, SystemTime, UNIX_EPOCH}; const VECTOR_ENGINE_SEND_MAX_ATTEMPTS: u32 = 5; const VECTOR_ENGINE_SEND_RETRY_BASE_DELAY_MS: u64 = 500; const VECTOR_ENGINE_SEND_RETRY_MAX_JITTER_MS: u64 = 999; use super::{ + budget::{ + effective_request_timeout_ms, request_budget_exhausted_error, + retry_delay_fits_request_deadline, + }, constants::{GPT_IMAGE_2_MODEL, VECTOR_ENGINE_PROVIDER}, curl_transport::{ map_curl_error, send_vector_engine_json_request_with_curl, @@ -63,8 +67,13 @@ pub async fn create_vector_engine_image_generation_with_model( ) -> Result { let model = normalize_vector_engine_image_model(model); if !reference_images.is_empty() { - let resolved_references = - resolve_reference_images(http_client, reference_images, failure_context).await?; + let resolved_references = resolve_reference_images( + http_client, + reference_images, + failure_context, + settings.request_deadline, + ) + .await?; return create_vector_engine_image_edit_with_references_and_model( http_client, settings, @@ -92,17 +101,28 @@ pub async fn create_vector_engine_image_generation_with_model( let started_at = std::time::Instant::now(); let mut attempt = 1; let response = loop { + let Some(attempt_timeout_ms) = + effective_request_timeout_ms(settings.request_timeout_ms, settings.request_deadline) + else { + return Err(request_budget_exhausted_error( + request_url.as_str(), + failure_context, + Some(started_at.elapsed().as_millis() as u64), + Some(prompt.chars().count()), + Some(reference_images.len()), + )); + }; match send_vector_engine_json_request_with_curl( request_url.as_str(), settings.api_key.as_str(), &request_body, - settings.request_timeout_ms, + attempt_timeout_ms, ) .await { Ok(response) => { if should_retry_vector_engine_upstream_status(response.status, attempt) { - retry_vector_engine_upstream_status_after_delay( + if retry_vector_engine_upstream_status_after_delay( "generation", request_url.as_str(), attempt, @@ -112,16 +132,19 @@ pub async fn create_vector_engine_image_generation_with_model( Some(prompt.chars().count()), Some(reference_images.len()), Some(&request_body), + settings.request_deadline, ) - .await; - attempt += 1; - continue; + .await + { + attempt += 1; + continue; + } } break response; } Err(error) => { if should_retry_vector_engine_curl_send_error(&error, attempt) { - retry_vector_engine_send_after_delay( + if retry_vector_engine_send_after_delay( "generation", request_url.as_str(), "request_send", @@ -135,10 +158,13 @@ pub async fn create_vector_engine_image_generation_with_model( Some(prompt.chars().count()), Some(reference_images.len()), Some(&request_body), + settings.request_deadline, ) - .await; - attempt += 1; - continue; + .await + { + attempt += 1; + continue; + } } return Err(map_curl_error( format!("{failure_context}：创建图片生成任务失败").as_str(), @@ -179,6 +205,7 @@ pub async fn create_vector_engine_image_generation_with_model( Some(reference_images.len()), candidate_count, "vector-engine", + settings.request_deadline, ) .await } @@ -227,17 +254,28 @@ pub async fn create_vector_engine_nanobanana_generate_content( let started_at = std::time::Instant::now(); let mut attempt = 1; let response = loop { + let Some(attempt_timeout_ms) = + effective_request_timeout_ms(settings.request_timeout_ms, settings.request_deadline) + else { + return Err(request_budget_exhausted_error( + request_url.as_str(), + failure_context, + Some(started_at.elapsed().as_millis() as u64), + Some(prompt.chars().count()), + Some(reference_image_count), + )); + }; match send_vector_engine_json_request_with_curl( request_url.as_str(), settings.api_key.as_str(), &request_body, - settings.request_timeout_ms, + attempt_timeout_ms, ) .await { Ok(response) => { if should_retry_vector_engine_upstream_status(response.status, attempt) { - retry_vector_engine_upstream_status_after_delay( + if retry_vector_engine_upstream_status_after_delay( "nanobanana_generate_content", request_url.as_str(), attempt, @@ -247,16 +285,19 @@ pub async fn create_vector_engine_nanobanana_generate_content( Some(prompt.chars().count()), Some(reference_image_count), Some(&request_params), + settings.request_deadline, ) - .await; - attempt += 1; - continue; + .await + { + attempt += 1; + continue; + } } break response; } Err(error) => { if should_retry_vector_engine_curl_send_error(&error, attempt) { - retry_vector_engine_send_after_delay( + if retry_vector_engine_send_after_delay( "nanobanana_generate_content", request_url.as_str(), "request_send", @@ -270,10 +311,13 @@ pub async fn create_vector_engine_nanobanana_generate_content( Some(prompt.chars().count()), Some(reference_image_count), Some(&request_params), + settings.request_deadline, ) - .await; - attempt += 1; - continue; + .await + { + attempt += 1; + continue; + } } return Err(map_curl_error( format!("{failure_context}：创建 nanobanana2 图片生成任务失败").as_str(), @@ -317,6 +361,7 @@ pub async fn create_vector_engine_nanobanana_generate_content( Some(reference_image_count), 1, "vector-engine-nanobanana", + settings.request_deadline, ) .await } @@ -427,6 +472,17 @@ pub async fn create_vector_engine_image_edit_with_references_and_model( ); let mut attempt = 1; let response = loop { + let Some(attempt_timeout_ms) = + effective_request_timeout_ms(settings.request_timeout_ms, settings.request_deadline) + else { + return Err(request_budget_exhausted_error( + request_url.as_str(), + failure_context, + Some(started_at.elapsed().as_millis() as u64), + Some(prompt.chars().count()), + Some(reference_image_count), + )); + }; match send_vector_engine_multipart_edit_request_with_curl( request_url.as_str(), settings.api_key.as_str(), @@ -436,13 +492,13 @@ pub async fn create_vector_engine_image_edit_with_references_and_model( normalized_size.as_str(), candidate_count, reference_images, - settings.request_timeout_ms, + attempt_timeout_ms, ) .await { Ok(response) => { if should_retry_vector_engine_upstream_status(response.status, attempt) { - retry_vector_engine_upstream_status_after_delay( + if retry_vector_engine_upstream_status_after_delay( "edit", request_url.as_str(), attempt, @@ -452,16 +508,19 @@ pub async fn create_vector_engine_image_edit_with_references_and_model( Some(prompt.chars().count()), Some(reference_image_count), Some(&request_params), + settings.request_deadline, ) - .await; - attempt += 1; - continue; + .await + { + attempt += 1; + continue; + } } break response; } Err(error) => { if should_retry_vector_engine_curl_send_error(&error, attempt) { - retry_vector_engine_send_after_delay( + if retry_vector_engine_send_after_delay( "edit", request_url.as_str(), "request_send", @@ -475,10 +534,13 @@ pub async fn create_vector_engine_image_edit_with_references_and_model( Some(prompt.chars().count()), Some(reference_image_count), Some(&request_params), + settings.request_deadline, ) - .await; - attempt += 1; - continue; + .await + { + attempt += 1; + continue; + } } return Err(map_curl_error( format!("{failure_context}：创建图片编辑任务失败").as_str(), @@ -520,6 +582,7 @@ pub async fn create_vector_engine_image_edit_with_references_and_model( Some(reference_image_count), candidate_count, "vector-engine-edit", + settings.request_deadline, ) .await } @@ -550,8 +613,31 @@ async fn retry_vector_engine_send_after_delay( prompt_chars: Option, reference_image_count: Option, request_params: Option<&serde_json::Value>, -) { + request_deadline: Option, +) -> bool { let delay_ms = vector_engine_send_retry_delay_ms(attempt, vector_engine_send_retry_jitter_ms()); + if !retry_delay_fits_request_deadline(request_deadline, delay_ms) { + tracing::warn!( + provider = VECTOR_ENGINE_PROVIDER, + endpoint = %request_url, + request_kind, + failure_stage = "request_budget", + attempt, + max_attempts = VECTOR_ENGINE_SEND_MAX_ATTEMPTS, + retry_delay_ms = delay_ms, + timeout, + connect, + request, + body, + status = 0, + error, + elapsed_ms, + prompt_chars, + reference_image_count, + "VectorEngine 图片请求剩余预算不足，停止重试" + ); + return false; + } tracing::warn!( provider = VECTOR_ENGINE_PROVIDER, endpoint = %request_url, @@ -575,6 +661,7 @@ async fn retry_vector_engine_send_after_delay( "VectorEngine 图片请求发送失败，准备重试" ); tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; + true } async fn retry_vector_engine_upstream_status_after_delay( @@ -587,8 +674,27 @@ async fn retry_vector_engine_upstream_status_after_delay( prompt_chars: Option, reference_image_count: Option, request_params: Option<&serde_json::Value>, -) { + request_deadline: Option, +) -> bool { let delay_ms = vector_engine_send_retry_delay_ms(attempt, vector_engine_send_retry_jitter_ms()); + if !retry_delay_fits_request_deadline(request_deadline, delay_ms) { + tracing::warn!( + provider = VECTOR_ENGINE_PROVIDER, + endpoint = %request_url, + request_kind, + failure_stage = "request_budget", + attempt, + max_attempts = VECTOR_ENGINE_SEND_MAX_ATTEMPTS, + retry_delay_ms = delay_ms, + status, + retryable = false, + elapsed_ms, + prompt_chars, + reference_image_count, + "VectorEngine 图片请求剩余预算不足，停止上游状态重试" + ); + return false; + } tracing::warn!( provider = VECTOR_ENGINE_PROVIDER, endpoint = %request_url, @@ -609,6 +715,7 @@ async fn retry_vector_engine_upstream_status_after_delay( "VectorEngine 图片上游状态可重试，准备重试" ); tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; + true } fn vector_engine_send_retry_delay_ms(attempt: u32, jitter_ms: u64) -> u64 { @@ -629,6 +736,40 @@ fn vector_engine_send_retry_jitter_ms() -> u64 { mod tests { use super::*; + #[tokio::test] + async fn expired_deadline_stops_generation_before_network_send() { + let settings = VectorEngineImageSettings { + base_url: "http://127.0.0.1:9".to_string(), + api_key: "test-key".to_string(), + request_timeout_ms: 1_000, + request_deadline: Some(Instant::now()), + }; + let error = create_vector_engine_image_generation( + &reqwest::Client::new(), + &settings, + "测试提示词", + None, + "1024x1024", + 1, + &[], + "测试图片生成失败", + ) + .await + .expect_err("expired deadline should stop before network send"); + + assert!(matches!( + error, + PlatformImageError::Request { timeout: true, .. } + )); + assert_eq!( + error + .audit() + .expect("budget error should carry audit") + .failure_stage, + "request_budget" + ); + } + #[test] fn vector_engine_send_retry_policy_allows_four_retries_before_final_attempt() { assert_eq!(VECTOR_ENGINE_SEND_MAX_ATTEMPTS, 5); diff --git a/server-rs/crates/platform-image/src/vector_engine/image_source.rs b/server-rs/crates/platform-image/src/vector_engine/image_source.rs index f04decf56..e2a59706c 100644 --- a/server-rs/crates/platform-image/src/vector_engine/image_source.rs +++ b/server-rs/crates/platform-image/src/vector_engine/image_source.rs @@ -1,7 +1,9 @@ use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD}; use reqwest::header; +use std::time::Instant; use super::{ + budget::request_budget_exhausted_error, constants::VECTOR_ENGINE_PROVIDER, error::PlatformImageError, types::{DownloadedImage, GeneratedImages, ReferenceImage}, @@ -53,18 +55,63 @@ pub async fn download_remote_image( }) } +async fn download_remote_image_with_deadline( + http_client: &reqwest::Client, + image_url: &str, + request_deadline: Option, + operation: &str, +) -> Result { + let Some(request_deadline) = request_deadline else { + return download_remote_image(http_client, image_url).await; + }; + let started_at = Instant::now(); + if request_deadline <= started_at { + return Err(request_budget_exhausted_error( + image_url, + operation, + Some(0), + None, + None, + )); + } + tokio::time::timeout_at( + tokio::time::Instant::from_std(request_deadline), + download_remote_image(http_client, image_url), + ) + .await + .map_err(|_| { + request_budget_exhausted_error( + image_url, + operation, + Some(started_at.elapsed().as_millis() as u64), + None, + None, + ) + })? +} + pub(crate) async fn download_images_from_urls( http_client: &reqwest::Client, task_id: String, image_urls: Vec, candidate_count: u32, + request_deadline: Option, + operation: &str, ) -> Result { let mut images = Vec::with_capacity(candidate_count.clamp(1, 4) as usize); for image_url in image_urls .into_iter() .take(candidate_count.clamp(1, 4) as usize) { - images.push(download_remote_image(http_client, image_url.as_str()).await?); + images.push( + download_remote_image_with_deadline( + http_client, + image_url.as_str(), + request_deadline, + operation, + ) + .await?, + ); } Ok(GeneratedImages { task_id, @@ -77,6 +124,7 @@ pub(crate) async fn resolve_reference_images( http_client: &reqwest::Client, reference_images: &[String], failure_context: &str, + request_deadline: Option, ) -> Result, PlatformImageError> { let mut resolved = Vec::new(); for (index, source) in reference_images.iter().take(5).enumerate() { @@ -89,20 +137,14 @@ pub(crate) async fn resolve_reference_images( continue; } if source.starts_with("http://") || source.starts_with("https://") { - let downloaded = download_remote_image(http_client, source) - .await - .map_err(|error| PlatformImageError::Request { - provider: VECTOR_ENGINE_PROVIDER, - message: format!("{failure_context}：下载参考图失败：{error}"), - endpoint: Some(source.to_string()), - timeout: false, - connect: false, - request: false, - body: false, - status_code: None, - source: None, - audit: None, - })?; + let downloaded = download_remote_image_with_deadline( + http_client, + source, + request_deadline, + failure_context, + ) + .await + .map_err(|error| contextualize_reference_download_error(error, failure_context))?; resolved.push(ReferenceImage { bytes: downloaded.bytes, mime_type: downloaded.mime_type.clone(), @@ -246,3 +288,106 @@ fn map_simple_request_error(message: String, endpoint: Option) -> Platfo audit: None, } } + +fn contextualize_reference_download_error( + error: PlatformImageError, + failure_context: &str, +) -> PlatformImageError { + match error { + PlatformImageError::Request { + provider, + message, + endpoint, + timeout, + connect, + request, + body, + status_code, + source, + audit, + } => PlatformImageError::Request { + provider, + message: format!("{failure_context}：下载参考图失败：{message}"), + endpoint, + timeout, + connect, + request, + body, + status_code, + source, + audit, + }, + error => error, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + use tokio::net::TcpListener; + + #[tokio::test] + async fn exhausted_deadline_stops_remote_image_download_before_send() { + let http_client = reqwest::Client::new(); + let error = download_remote_image_with_deadline( + &http_client, + "http://127.0.0.1:9/not-called.png", + Some(Instant::now()), + "下载测试图片失败", + ) + .await + .expect_err("expired deadline should stop before network send"); + + assert!(matches!( + error, + PlatformImageError::Request { timeout: true, .. } + )); + assert_eq!( + error + .audit() + .expect("budget error should carry audit") + .failure_stage, + "request_budget" + ); + } + + #[tokio::test] + async fn deadline_cancels_pending_remote_image_download() { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("mock listener should bind"); + let address = listener + .local_addr() + .expect("mock listener should expose address"); + let server = tokio::spawn(async move { + let (_socket, _) = listener + .accept() + .await + .expect("mock listener should accept request"); + tokio::time::sleep(Duration::from_secs(1)).await; + }); + + let error = download_remote_image_with_deadline( + &reqwest::Client::new(), + format!("http://{address}/pending.png").as_str(), + Some(Instant::now() + Duration::from_millis(100)), + "下载测试图片失败", + ) + .await + .expect_err("absolute deadline should cancel pending download"); + server.abort(); + + assert!(matches!( + error, + PlatformImageError::Request { timeout: true, .. } + )); + assert_eq!( + error + .audit() + .expect("budget error should carry audit") + .failure_stage, + "request_budget" + ); + } +} diff --git a/server-rs/crates/platform-image/src/vector_engine/mod.rs b/server-rs/crates/platform-image/src/vector_engine/mod.rs index 1422ecb3a..8f771c54a 100644 --- a/server-rs/crates/platform-image/src/vector_engine/mod.rs +++ b/server-rs/crates/platform-image/src/vector_engine/mod.rs @@ -1,4 +1,5 @@ mod audit; +mod budget; mod client; mod constants; mod curl_transport; diff --git a/server-rs/crates/platform-image/src/vector_engine/response.rs b/server-rs/crates/platform-image/src/vector_engine/response.rs index 2d98918e7..112889930 100644 --- a/server-rs/crates/platform-image/src/vector_engine/response.rs +++ b/server-rs/crates/platform-image/src/vector_engine/response.rs @@ -10,6 +10,7 @@ use super::{ types::GeneratedImages, util::{current_utc_micros, is_timeout_message, truncate_raw}, }; +use std::time::Instant; pub(crate) async fn handle_vector_engine_response( http_client: &reqwest::Client, @@ -22,6 +23,7 @@ pub(crate) async fn handle_vector_engine_response( reference_image_count: Option, candidate_count: u32, task_prefix: &str, + request_deadline: Option, ) -> Result { if !(200..=299).contains(&response_status) { let message = parse_api_error_message(response_text, failure_context); @@ -101,18 +103,32 @@ pub(crate) async fn handle_vector_engine_response( task_id, image_urls, candidate_count, + request_deadline, + failure_context, ) .await { Ok(generated) => generated, Err(error) => { + let timeout = matches!(&error, PlatformImageError::Request { timeout: true, .. }); + let request_budget_exhausted = error + .audit() + .is_some_and(|audit| audit.failure_stage == "request_budget"); let audit = build_failure_audit( request_url, failure_context, - "image_download", + if request_budget_exhausted { + "request_budget" + } else { + "image_download" + }, Some(response_status), - Some("5xx"), - false, + if request_budget_exhausted { + None + } else { + Some("5xx") + }, + timeout, false, error.message(), None, diff --git a/server-rs/crates/platform-image/src/vector_engine/tests.rs b/server-rs/crates/platform-image/src/vector_engine/tests.rs index dc9a37ea6..febcd539f 100644 --- a/server-rs/crates/platform-image/src/vector_engine/tests.rs +++ b/server-rs/crates/platform-image/src/vector_engine/tests.rs @@ -27,11 +27,13 @@ mod tests { base_url: "https://vector.example".to_string(), api_key: "test-key".to_string(), request_timeout_ms: 1_000, + request_deadline: None, }; let v1_settings = VectorEngineImageSettings { base_url: "https://vector.example/v1".to_string(), api_key: "test-key".to_string(), request_timeout_ms: 1_000, + request_deadline: None, }; assert_eq!( diff --git a/server-rs/crates/platform-image/src/vector_engine/types.rs b/server-rs/crates/platform-image/src/vector_engine/types.rs index 5228b04b5..328173567 100644 --- a/server-rs/crates/platform-image/src/vector_engine/types.rs +++ b/server-rs/crates/platform-image/src/vector_engine/types.rs @@ -3,6 +3,7 @@ pub struct VectorEngineImageSettings { pub base_url: String, pub api_key: String, pub request_timeout_ms: u64, + pub request_deadline: Option, } #[derive(Clone, Debug)] diff --git a/server-rs/crates/platform-image/tests/vector_engine.rs b/server-rs/crates/platform-image/tests/vector_engine.rs index 0ec6d9c43..6485e8d6c 100644 --- a/server-rs/crates/platform-image/tests/vector_engine.rs +++ b/server-rs/crates/platform-image/tests/vector_engine.rs @@ -1,7 +1,7 @@ use platform_image::vector_engine::{ - GPT_IMAGE_2_MODEL, ReferenceImage, VECTOR_ENGINE_PROVIDER, VectorEngineImageSettings, - build_vector_engine_image_http_client, build_vector_engine_image_request_body, - build_vector_engine_image_request_body_with_model, + GPT_IMAGE_2_MODEL, PlatformImageError, ReferenceImage, VECTOR_ENGINE_PROVIDER, + VectorEngineImageSettings, build_vector_engine_image_http_client, + build_vector_engine_image_request_body, build_vector_engine_image_request_body_with_model, build_vector_engine_nanobanana_generate_content_request_body, create_vector_engine_image_edit, create_vector_engine_image_generation, create_vector_engine_nanobanana_generate_content, vector_engine_images_edit_url, vector_engine_images_generation_url, @@ -12,7 +12,7 @@ use std::{ Arc, atomic::{AtomicUsize, Ordering}, }, - time::Duration, + time::{Duration, Instant}, }; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, @@ -25,6 +25,7 @@ fn vector_engine_module_exposes_provider_protocol_helpers() { base_url: "https://vector.example/v1".to_string(), api_key: "test-key".to_string(), request_timeout_ms: 1_000, + request_deadline: None, }; let body = @@ -187,6 +188,7 @@ fn nanobanana_generate_content_url_uses_model_path() { base_url: "https://vector.example/v1".to_string(), api_key: "test-key".to_string(), request_timeout_ms: 1_000, + request_deadline: None, }; assert_eq!( @@ -235,6 +237,7 @@ async fn vector_engine_image_edit_retries_send_timeout_once_and_succeeds() { base_url: format!("http://{server_addr}/v1"), api_key: "test-key".to_string(), request_timeout_ms: 40, + request_deadline: None, }; let http_client = build_vector_engine_image_http_client(&settings).expect("client should build"); @@ -262,6 +265,65 @@ async fn vector_engine_image_edit_retries_send_timeout_once_and_succeeds() { server.abort(); } +#[tokio::test] +async fn vector_engine_deadline_clips_stalled_attempt_and_prevents_retry() { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("mock server should bind"); + let server_addr = listener + .local_addr() + .expect("mock server address should be readable"); + let request_count = Arc::new(AtomicUsize::new(0)); + let request_count_for_server = Arc::clone(&request_count); + let server = tokio::spawn(async move { + loop { + let Ok((mut stream, _)) = listener.accept().await else { + break; + }; + request_count_for_server.fetch_add(1, Ordering::SeqCst); + tokio::spawn(async move { + let mut buffer = [0_u8; 4096]; + let _ = stream.read(&mut buffer).await; + tokio::time::sleep(Duration::from_secs(1)).await; + }); + } + }); + + let started_at = Instant::now(); + let settings = VectorEngineImageSettings { + base_url: format!("http://{server_addr}/v1"), + api_key: "test-key".to_string(), + request_timeout_ms: 5_000, + request_deadline: Some(started_at + Duration::from_millis(150)), + }; + let http_client = + build_vector_engine_image_http_client(&settings).expect("client should build"); + + let error = create_vector_engine_image_generation( + &http_client, + &settings, + "测试提示词", + None, + "1024x1024", + 1, + &[], + "测试 VectorEngine 图片生成失败", + ) + .await + .expect_err("stalled request should exhaust the shared deadline"); + + assert!(matches!( + error, + PlatformImageError::Request { timeout: true, .. } + )); + assert!( + started_at.elapsed() < Duration::from_secs(1), + "attempt 应使用剩余 deadline，而不是完整配置 timeout" + ); + assert_eq!(request_count.load(Ordering::SeqCst), 1); + server.abort(); +} + #[tokio::test] async fn nanobanana_generate_content_posts_native_body_and_reads_inline_data() { let listener = TcpListener::bind("127.0.0.1:0") @@ -307,6 +369,7 @@ async fn nanobanana_generate_content_posts_native_body_and_reads_inline_data() { base_url: format!("http://{}", server_addr), api_key: "test-key".to_string(), request_timeout_ms: 1_000, + request_deadline: None, }; let client = build_vector_engine_image_http_client(&settings).expect("client should build"); @@ -375,6 +438,7 @@ async fn vector_engine_image_generation_retries_upstream_502_once_and_succeeds() base_url: format!("http://{server_addr}/v1"), api_key: "test-key".to_string(), request_timeout_ms: 1_000, + request_deadline: None, }; let http_client = build_vector_engine_image_http_client(&settings).expect("client should build");