diff --git a/docs/technical/【后端架构】外部生成Worker化方案-2026-06-03.md b/docs/technical/【后端架构】外部生成Worker化方案-2026-06-03.md index ed169435b..5686c2093 100644 --- a/docs/technical/【后端架构】外部生成Worker化方案-2026-06-03.md +++ b/docs/technical/【后端架构】外部生成Worker化方案-2026-06-03.md @@ -137,7 +137,7 @@ worker 配置： - `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`：VectorEngine 图片生成 / 编辑、图标 spritesheet 生成、UI 素材提取以及角色动作、视频等长耗时 job 的执行预算，默认 `1800`。其中四类 VectorEngine 图片 job 固定为 `editor_image_generation`、`editor_image_edit`、`editor_icon_spritesheet_generation` 和 `editor_ui_design_asset_extraction`；手动去背景等不直接调用 VectorEngine 的 job 继续使用普通预算。Tripo 3D 的 `model3d_text_to_model` 与 `model3d_image_to_model` 同样使用该预算：它们在单次 attempt 内把 `get_task` 轮询到 provider 终态再下载产物，因此会持续占用一个 worker 并发位，接入真实流量前必须确认并发数与超时预算。 -- `TRIPO_BASE_URL` / `TRIPO_API_KEY` / `TRIPO_REQUEST_TIMEOUT_MS` / `TRIPO_RETRIES`：Tripo 3D provider 的网关、凭据、单请求超时与重试次数，由 HTTP 角色与 worker 共享同一份 API env。默认网关 `https://openapi.tripo3d.com/v3`；缺 API Key 时 3D 提交返回 503 且不扣费、不入队。worker 只使用 submit、单次 `get_task` 与产物下载三项能力，任务 ID 只作为 checkpoint 存在服务端。 +- `TRIPO_BASE_URL` / `TRIPO_API_KEY` / `TRIPO_REQUEST_TIMEOUT_MS` / `TRIPO_RETRIES`：Tripo 3D provider 的网关、凭据、API 单请求超时（同时也是产物下载的连接与「无数据推进」预算）与重试次数，由 HTTP 角色与 worker 共享同一份 API env。默认网关 `https://openapi.tripo3d.com/v3`；缺 API Key 时 3D 提交返回 503 且不扣费、不入队。worker 只使用 submit、单次 `get_task` 与产物下载三项能力，任务 ID 只作为 checkpoint 存在服务端。 worker 在单次 job 开始执行时从同一个单调时钟起点计算绝对 `job deadline` 和更早的 `provider deadline`：常规情况下为终态审计、OSS 持久化及 `complete/fail` 回写保留 `60` 秒；当整个 job 预算小于 `120` 秒时，保留其一半，避免 provider 预算被全部吃掉。该 deadline 只通过进程内 `RequestContext` 传给 VectorEngine 图片调用，不写入 HTTP DTO、队列 payload 或 SpacetimeDB；普通 HTTP / `inline` 上下文没有 deadline，保持原有行为。 diff --git a/docs/technical/【技术方案】Tripo 3D模型Provider集成-2026-09-18.md b/docs/technical/【技术方案】Tripo 3D模型Provider集成-2026-09-18.md index 52b906e14..b42e3414e 100644 --- a/docs/technical/【技术方案】Tripo 3D模型Provider集成-2026-09-18.md +++ b/docs/technical/【技术方案】Tripo 3D模型Provider集成-2026-09-18.md @@ -42,7 +42,7 @@ task 尚未完成时 `output` 为空；完成后 `output` 必须是与 task type `get_task` 按 `TripoTaskHandle` 做单次查询，返回通用 `TripoTaskSnapshot`，由 `task_type` 决定 `output` 的具体 variant；adapter 不做轮询、不阻塞等待。 -`download_model` 接受 `TripoTaskSnapshot`，先确认快照确实处于完成态，再从严格 endpoint 结果中取得已校验的 `model_url`，由 provider 自己的无鉴权 reqwest client 打开签名 URL 并返回 `TripoDownloadedArtifact` 流包装。包装只公开 `url`、`content_type`、`content_length`、`filename(name)` 和 `next_chunk()`；不提供完整 `Vec`，调用方必须逐块消费。`filename(name)` 的扩展名来自远端地址，只接受短的 ASCII 字母数字，其余退回 `glb`，避免远端地址里的 `%2F` 解码后拼出跨目录路径。provider 在流结束时校验实际接收字节数与 `Content-Length`，响应体中断或长度不一致按结构化请求错误失败；SDK 保持第三方原样，不承担产物下载。smoke example 使用异步文件写入逐块落盘；未来接入 OSS 时应把同一数据流直接送入 OSS 分片上传，不经过完整内存缓冲。 +`download_model` 接受 `TripoTaskSnapshot`，先确认快照确实处于完成态，再从严格 endpoint 结果中取得已校验的 `model_url`，由 provider 自己的无鉴权 reqwest client 打开签名 URL 并返回 `TripoDownloadedArtifact` 流包装。包装只公开 `url`、`content_type`、`content_length`、`filename(name)` 和 `next_chunk()`；不提供完整 `Vec`，调用方必须逐块消费。`filename(name)` 的扩展名来自远端地址，只接受短的 ASCII 字母数字，其余退回 `glb`，避免远端地址里的 `%2F` 解码后拼出跨目录路径。provider 在流结束时校验实际接收字节数与 `Content-Length`，响应体中断或长度不一致按结构化请求错误失败。产物下载不设总超时：几十 MB 的流只要还在出数据就不该被判失败、更不该重试后从零重传，因此下载链路只按「无数据推进」判超时，预算取 `TripoSettings::request_timeout`（连接用 `connect_timeout`，响应体用 `read_timeout`，响应头之前由显式的 `tokio::time::timeout` 兜住），超过预算没有数据推进即按传输失败返回；SDK 保持第三方原样，不承担产物下载。smoke example 使用异步文件写入逐块落盘；未来接入 OSS 时应把同一数据流直接送入 OSS 分片上传，不经过完整内存缓冲。 TODO：等待上游 `tripo-rust-sdk` 提供原生 artifact stream API 后，删除 provider-side reqwest 下载器，改由 SDK stream 直接承接。 @@ -53,4 +53,5 @@ TODO：等待上游 `tripo-rust-sdk` 提供原生 artifact stream API 后，删 - 三个入口的输入校验可拒绝空白 `prompt` / `input`、视图不足两张和空白 `taskId`；组合校验在调用 SDK 前完成。 - provider 错误统一为 adapter 错误类型。 - shared contracts 的 ts-rs binding 无 feature 开关且始终可生成。 +- 产物下载只按「无数据推进」判超时：连接、响应头与每个数据块都必须有进展，慢速但持续的下载可完整收完，中途断流按请求错误失败。 - 真实 Provider smoke 已覆盖三个入口；example 在任务未完成时跳过下载并继续跑后续入口，只打印脱敏后的结果结构和下载产物信息。 diff --git a/server-rs/crates/platform-tripo/Cargo.toml b/server-rs/crates/platform-tripo/Cargo.toml index 086a9fec8..a8fe432a5 100644 --- a/server-rs/crates/platform-tripo/Cargo.toml +++ b/server-rs/crates/platform-tripo/Cargo.toml @@ -15,4 +15,4 @@ tokio = { workspace = true, features = ["time"] } url = { workspace = true } [dev-dependencies] -tokio = { workspace = true, features = ["fs", "io-util", "macros", "rt-multi-thread"] } +tokio = { workspace = true, features = ["fs", "io-util", "macros", "net", "rt-multi-thread"] } diff --git a/server-rs/crates/platform-tripo/src/common/client.rs b/server-rs/crates/platform-tripo/src/common/client.rs index 5c31a5bcc..d08f25ff9 100644 --- a/server-rs/crates/platform-tripo/src/common/client.rs +++ b/server-rs/crates/platform-tripo/src/common/client.rs @@ -12,13 +12,19 @@ pub struct TripoProviderClient { pub(crate) client: TripoClient, artifact_client: reqwest::Client, artifact_retries: u32, + /// 产物下载的「无数据推进」预算：连接、首字节和之后每个数据块都必须在它之内有进展， + /// 下载总时长不设限。 + artifact_stall_timeout: Duration, } impl TripoProviderClient { pub fn new(settings: TripoSettings) -> Result { let artifact_client = reqwest::Client::builder() .user_agent(settings.user_agent.clone()) - .timeout(settings.request_timeout) + // 产物是几十 MB 的流，设总超时会把「还在正常下载」判成失败，也会让重试从头重传； + // 这里只限制「多久没有进展」：连接用 connect_timeout，传输过程用 read_timeout。 + .connect_timeout(settings.request_timeout) + .read_timeout(settings.request_timeout) .build() .map_err(|error| TripoError::Request { message: format!("failed to build artifact download client: {error}"), @@ -28,6 +34,7 @@ impl TripoProviderClient { client: TripoClient::new(settings.client_options()).map_err(TripoError::from)?, artifact_client, artifact_retries: settings.retries, + artifact_stall_timeout: settings.request_timeout, }) } @@ -93,18 +100,21 @@ impl TripoProviderClient { task_id: &str, url: &TripoUrl, ) -> Result { - let response = self - .artifact_client - .get(url.as_str()) - .send() - .await - .map_err(|error| TripoError::Request { - message: format!( - "artifact download transport failure for task {task_id}: {}", - transport_error_kind(&error) - ), - status: None, - })?; + // read_timeout 只覆盖响应体，响应头之前没有进展同样要按「无数据推进」判超时， + // 否则连上却不给响应头的服务端会让下载无限挂住。 + let response = tokio::time::timeout( + self.artifact_stall_timeout, + self.artifact_client.get(url.as_str()).send(), + ) + .await + .map_err(|_elapsed| artifact_stall_error(task_id, self.artifact_stall_timeout))? + .map_err(|error| TripoError::Request { + message: format!( + "artifact download transport failure for task {task_id}: {}", + transport_error_kind(&error) + ), + status: None, + })?; let status = response.status(); if !status.is_success() { return Err(TripoError::Request { @@ -132,6 +142,16 @@ fn download_backoff(attempt: u32) -> Duration { Duration::from_millis(250u64.saturating_mul(2u64.saturating_pow(attempt.min(6)))) } +/// 产物下载长时间没有数据推进：连接、响应头或响应体任一段卡住都归一成这个错误。 +fn artifact_stall_error(task_id: &str, stall_timeout: Duration) -> TripoError { + TripoError::Request { + message: format!( + "artifact download stalled for task {task_id}: no data received within {stall_timeout:?}" + ), + status: None, + } +} + /// 取完成态产物：下载入口只接受已完成且带输出的 task 快照。 /// /// `TripoTaskSnapshot` 字段全部公开，「完成态才有 output」只是 `map_task` 的约定， @@ -162,3 +182,164 @@ fn transport_error_kind(error: &reqwest::Error) -> &'static str { "request error" } } + +#[cfg(test)] +mod tests { + use std::net::SocketAddr; + use std::time::Duration; + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + use super::*; + + /// 测试用的「无数据推进」预算：够短让用例跑得快，又给 CI 抖动留出余量。 + const STALL_TIMEOUT: Duration = Duration::from_millis(300); + /// 正常推进的间隔：明显小于预算，保证用例只在「真的没数据」时才失败。 + const PROGRESS_INTERVAL: Duration = Duration::from_millis(100); + + fn test_client() -> TripoProviderClient { + let settings = TripoSettings::new( + "test-key".to_string(), + "http://127.0.0.1:1".to_string(), + STALL_TIMEOUT, + 0, + "genarrative-test-tripo/1".to_string(), + ); + TripoProviderClient::new(settings).expect("测试配置必须能建出 client") + } + + fn artifact_url(addr: SocketAddr) -> TripoUrl { + TripoUrl::parse(&format!("http://{addr}/model.glb")).expect("测试地址必须是合法产物 URL") + } + + async fn spawn_mock_server(serve: F) -> SocketAddr + where + F: FnOnce(tokio::net::TcpStream) -> Fut + Send + 'static, + Fut: std::future::Future + Send, + { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("mock server should bind"); + let addr = listener + .local_addr() + .expect("mock server should expose addr"); + tokio::spawn(async move { + let Ok((socket, _)) = listener.accept().await else { + return; + }; + serve(socket).await; + }); + addr + } + + async fn read_request(socket: &mut tokio::net::TcpStream) { + let mut buffer = [0u8; 1024]; + let _ = socket.read(&mut buffer).await; + } + + /// 接受连接后一直不返回响应头，直到远超预算才断开。 + async fn spawn_silent_server() -> SocketAddr { + spawn_mock_server(|_socket| async { + tokio::time::sleep(Duration::from_secs(30)).await; + }) + .await + } + + /// 响应头与第一块数据都正常，之后彻底断流。 + async fn spawn_stalling_body_server() -> SocketAddr { + spawn_mock_server(|mut socket| async move { + read_request(&mut socket).await; + let head = "HTTP/1.1 200 OK\r\n\ + Content-Type: model/gltf-binary\r\n\ + Content-Length: 64\r\n\ + \r\n"; + if socket.write_all(head.as_bytes()).await.is_err() { + return; + } + if socket.write_all(b"glTF").await.is_err() { + return; + } + let _ = socket.flush().await; + tokio::time::sleep(Duration::from_secs(30)).await; + }) + .await + } + + /// 总量固定但推得很慢：总时长远超预算，每次却都有数据推进。 + async fn spawn_slow_progress_server() -> SocketAddr { + spawn_mock_server(|mut socket| async move { + let _ = socket.set_nodelay(true); + read_request(&mut socket).await; + let head = "HTTP/1.1 200 OK\r\n\ + Content-Type: model/gltf-binary\r\n\ + Content-Length: 10\r\n\ + \r\n"; + if socket.write_all(head.as_bytes()).await.is_err() { + return; + } + for _ in 0..5 { + tokio::time::sleep(PROGRESS_INTERVAL).await; + if socket.write_all(b"gl").await.is_err() { + return; + } + let _ = socket.flush().await; + } + }) + .await + } + + #[tokio::test] + async fn artifact_download_times_out_when_response_headers_never_arrive() { + let addr = spawn_silent_server().await; + let error = match test_client() + .download_artifact("task-1", &artifact_url(addr)) + .await + { + Ok(_) => panic!("响应头一直不来必须按无数据推进失败"), + Err(error) => error, + }; + assert!(matches!(error, TripoError::Request { .. }), "{error:?}"); + } + + #[tokio::test] + async fn artifact_download_times_out_when_body_stops_progressing() { + let addr = spawn_stalling_body_server().await; + let mut artifact = test_client() + .download_artifact_once("task-1", &artifact_url(addr)) + .await + .expect("响应头与第一块数据必须正常返回"); + + let first = artifact + .next_chunk() + .await + .expect("第一块数据必须正常返回") + .expect("第一块数据必须存在"); + assert_eq!(first.len(), 4); + + let error = artifact + .next_chunk() + .await + .expect_err("传输中途断流必须按无数据推进失败"); + assert!(matches!(error, TripoError::Request { .. }), "{error:?}"); + } + + #[tokio::test] + async fn artifact_download_survives_slow_but_progressing_body() { + let addr = spawn_slow_progress_server().await; + let mut artifact = test_client() + .download_artifact_once("task-1", &artifact_url(addr)) + .await + .expect("响应头必须正常返回"); + + let mut received = Vec::new(); + while let Some(chunk) = artifact + .next_chunk() + .await + .expect("只要还在出数据就不能判超时") + { + received.extend_from_slice(&chunk); + } + assert_eq!(received.len(), 10, "慢但持续的下载必须完整收完"); + } +}