From b33a8a141cafed851e8af5cf1331d4388d260052 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E5=BE=B7=E5=AE=87?= Date: Wed, 23 Sep 2026 17:01:40 +0800 Subject: [PATCH] =?UTF-8?q?3D=20=E4=BA=A7=E7=89=A9=E4=B8=8B=E8=BD=BD?= =?UTF-8?q?=E7=9A=84=E9=87=8D=E8=AF=95=E6=94=B9=E4=B8=BA=E8=A6=86=E7=9B=96?= =?UTF-8?q?=E6=95=B4=E4=B8=AA=E5=93=8D=E5=BA=94=E4=BD=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - download_model / download_rendered_image 把「取响应头 + 读完 body」作为一次可重试事务， 每次尝试都重新取响应头并整体重下，不做断点续传 - body 传输失败与长度不一致归一成可重试的传输错误（不再带 HTTP 200），HTTP 状态只留在文案里 - 新增 TripoArtifactBytes（读完的产物）并公开 download_model / download_rendered_image 的 max_bytes 上限； 超限按 OutputSchema 失败且不重试，流式句柄 TripoDownloadedArtifact 收回 crate 内部 - api-server 的 read_artifact 改成对完整产物做落库前校验（只判空），体积上限转由调用方传入 - smoke 示例改为一次写入完整字节；新增「body 中途断开后整体重下成功」与「超限不重试」两个用例 - 同步技术方案与实施计划：把「只重试响应头」的描述改成「重试包住整个 body」 --- ...Tripo生成Worker执行链路与API路由-2026-09-21.md | 2 +- ...方案】Tripo 3D模型Provider集成-2026-09-18.md | 4 +- .../api-server/src/tripo3d/artifacts.rs | 48 ++----- .../crates/api-server/src/tripo3d/worker.rs | 12 +- .../examples/tripo_generation_smoke.rs | 23 +-- .../platform-tripo/src/common/client.rs | 130 +++++++++++++++-- .../crates/platform-tripo/src/common/mod.rs | 3 +- .../crates/platform-tripo/src/common/types.rs | 131 ++++++++++++++---- server-rs/crates/platform-tripo/src/lib.rs | 2 +- 9 files changed, 254 insertions(+), 101 deletions(-) diff --git a/docs/project-memory/plans/【实施计划】Tripo生成Worker执行链路与API路由-2026-09-21.md b/docs/project-memory/plans/【实施计划】Tripo生成Worker执行链路与API路由-2026-09-21.md index 4fb213163..6ae7b7da3 100644 --- a/docs/project-memory/plans/【实施计划】Tripo生成Worker执行链路与API路由-2026-09-21.md +++ b/docs/project-memory/plans/【实施计划】Tripo生成Worker执行链路与API路由-2026-09-21.md @@ -49,7 +49,7 @@ Milestone Spec: `docs/project-memory/plans/【里程碑】Tripo生成Worker执 - 验收：跨 owner 的资源 / 素材被拒绝。 5. **产物落库** - - provider 产物下载用 `next_chunk` 收进 `Vec`（流式直传 OSS 留 TODO），模型与预览分别 OSS 写入并记录 `content_type` / `content_length` / `sha256`。 + - provider 产物下载由 `platform-tripo` 的下载入口读完整个 body（body 中途失败整体重下，体积上限由调用方传入），模型与预览分别 OSS 写入并记录 `content_type` / `content_length` / `sha256`（流式直传 OSS 留 TODO）。 - 复用现有原子落库路径写入 `assetKind = "model3d"` 的资源 / 素材、预览图引用与 job 终态；不新增资源列。 - 按「新增编辑器 `assetKind` 接入清单」对照表执行：把两个 Tripo job kind 加进 `EDITOR_GENERATION_OPERATION_KINDS`（否则原子落库整笔事务被拒）；画布回填只走 placement-only，`model3d` 不进 `EDITOR_CANVAS_ASSET_KINDS`，也不动前端 `CANVAS_ASSET_KIND_TAG_OPTIONS`。 - 验收：两个 job kind 在白名单内；`canvasCompletion` 回填的 layer 只有 `layerId` / `resourceId`，无 `assetKind` / `src` / `objectKey`；用户标签覆盖、快速编辑、改造 capability 三项保持不进入。 diff --git a/docs/technical/【技术方案】Tripo 3D模型Provider集成-2026-09-18.md b/docs/technical/【技术方案】Tripo 3D模型Provider集成-2026-09-18.md index 12cb83457..d70bed68a 100644 --- a/docs/technical/【技术方案】Tripo 3D模型Provider集成-2026-09-18.md +++ b/docs/technical/【技术方案】Tripo 3D模型Provider集成-2026-09-18.md @@ -44,7 +44,7 @@ TODO：图生输入当前由 api-server 自己读站内对象的字节再上传 `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`，响应体中断或长度不一致按结构化请求错误失败。产物下载不设总超时：几十 MB 的流只要还在出数据就不该被判失败、更不该重试后从零重传，因此下载链路只按「无数据推进」判超时，预算取 `TripoSettings::request_timeout`（连接用 `connect_timeout`，响应体用 `read_timeout`，响应头之前由显式的 `tokio::time::timeout` 兜住），超过预算没有数据推进即按传输失败返回；SDK 保持第三方原样，不承担产物下载。smoke example 使用异步文件写入逐块落盘；未来接入 OSS 时应把同一数据流直接送入 OSS 分片上传，不经过完整内存缓冲。 +`download_model`（及预览图的 `download_rendered_image`）接受 `TripoTaskSnapshot` 与调用方给的体积上限 `max_bytes`，先确认快照确实处于完成态，再从严格 endpoint 结果中取得已校验的产物地址，由 provider 自己的无鉴权 reqwest client 打开签名 URL，读完整个 body 后返回 `TripoArtifactBytes`（`url`、`content_type`、`content_length`、`bytes`、`filename(name)`）。流式句柄 `TripoDownloadedArtifact`（`next_chunk()`）保留为内部实现：**重试必须包住整个 body** —— 只重试「拿到响应头」这一步的时候，几十 MB 的字节其实是在之后才传输的，一次 CDN 抖动就会毁掉一次已经扣费、provider 任务也跑完的生成。因此每次尝试都重新取响应头并整体重下（不做断点续传，签名地址会过期），body 读取失败与长度不一致都归一成可重试的传输错误，重试上限与退避沿用 `TripoSettings::retries`。`filename(name)` 的扩展名来自远端地址，只接受短的 ASCII 字母数字，其余退回 `glb`，避免远端地址里的 `%2F` 解码后拼出跨目录路径。每次读取都校验实际接收字节数与 `Content-Length`，读到 `max_bytes` 之上直接按输出违约失败（不重试）：宁可失败退款也不要把 api-server 内存打满。产物下载不设总超时：几十 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 直接承接。 @@ -55,5 +55,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/api-server/src/tripo3d/artifacts.rs b/server-rs/crates/api-server/src/tripo3d/artifacts.rs index 179455594..59238cf47 100644 --- a/server-rs/crates/api-server/src/tripo3d/artifacts.rs +++ b/server-rs/crates/api-server/src/tripo3d/artifacts.rs @@ -1,45 +1,31 @@ //! provider 产物的读取。 //! -//! provider 侧提供的是流式句柄，这里按字节收口成 `Vec` 交给 OSS 写入。 +//! 下载与「body 中途失败整体重下」都在 `platform-tripo` 的下载入口里完成（重试上限 +//! 取自 provider 客户端配置），这里只做 api-server 自己的口径：产物不能为空。 //! TODO(stream): provider SDK 与 `platform-oss` 都支持流式 / 分片后，这里应当直接 //! 把流转交 OSS，不再把几十 MB 的模型完整读进 api-server 内存。 use axum::http::StatusCode; -use platform_tripo::TripoDownloadedArtifact; +use platform_tripo::TripoArtifactBytes; use serde_json::json; use crate::http_error::AppError; -use super::{errors::map_provider_error, provider::TRIPO_PROVIDER}; +use super::provider::TRIPO_PROVIDER; -/// 单次 attempt 内允许读取的单个产物上限。超过它说明响应异常或产物口径变了， +/// 单次下载允许的单个产物上限。超过它说明响应异常或产物口径变了， /// 宁可失败退款也不要把 worker 的内存打满。 -const MODEL3D_MAX_ARTIFACT_BYTES: u64 = 512 * 1024 * 1024; +pub(crate) const MODEL3D_MAX_ARTIFACT_BYTES: u64 = 512 * 1024 * 1024; pub(crate) struct DownloadedArtifact { pub(crate) bytes: Vec, pub(crate) content_type: String, } -pub(crate) async fn read_artifact( - mut artifact: TripoDownloadedArtifact, -) -> Result { - let content_type = artifact.content_type.take().unwrap_or_default(); - // 按声明长度预分配（封顶到单产物上限，防止 provider 谎报长度撑爆内存）： - // 几百 MB 的模型逐块 append 会反复搬移，预分配把峰值与拷贝都压下来。 - let claimed_bytes = artifact - .content_length - .unwrap_or(0) - .min(MODEL3D_MAX_ARTIFACT_BYTES) as usize; - let mut bytes = Vec::with_capacity(claimed_bytes); - while let Some(chunk) = artifact.next_chunk().await.map_err(map_provider_error)? { - let next_len = bytes.len() as u64 + chunk.len() as u64; - if next_len > MODEL3D_MAX_ARTIFACT_BYTES { - return Err(artifact_too_large(next_len)); - } - bytes.extend_from_slice(chunk.as_ref()); - } - if bytes.is_empty() { +/// 完整读出的产物在落库前的最后一道检查：空产物按上游内容不合法失败， +/// 不写出一个 0 字节的对象再让下游去猜。 +pub(crate) fn read_artifact(artifact: TripoArtifactBytes) -> Result { + if artifact.bytes.is_empty() { return Err( AppError::from_status(StatusCode::BAD_GATEWAY).with_details(json!({ "provider": TRIPO_PROVIDER, @@ -49,17 +35,7 @@ pub(crate) async fn read_artifact( ); } Ok(DownloadedArtifact { - bytes, - content_type, + bytes: artifact.bytes, + content_type: artifact.content_type.unwrap_or_default(), }) } - -fn artifact_too_large(actual_bytes: u64) -> AppError { - AppError::from_status(StatusCode::BAD_GATEWAY).with_details(json!({ - "provider": TRIPO_PROVIDER, - "reason": "model3d-artifact-too-large", - "message": "provider 返回的 3D 产物超过单次尝试允许的体积上限。", - "actualBytes": actual_bytes, - "maxBytes": MODEL3D_MAX_ARTIFACT_BYTES, - })) -} diff --git a/server-rs/crates/api-server/src/tripo3d/worker.rs b/server-rs/crates/api-server/src/tripo3d/worker.rs index fd6adaa0f..1b280a8f4 100644 --- a/server-rs/crates/api-server/src/tripo3d/worker.rs +++ b/server-rs/crates/api-server/src/tripo3d/worker.rs @@ -33,7 +33,7 @@ use crate::{ }; use super::{ - artifacts::read_artifact, + artifacts::{MODEL3D_MAX_ARTIFACT_BYTES, read_artifact}, errors::map_provider_error, image_source::resolve_image_input, job::{MODEL3D_PROVIDER_KIND, Model3dJobRequest, parse_model3d_job_request}, @@ -85,18 +85,16 @@ async fn run_model3d_job( let snapshot = poll_until_terminal(&client, &handle, provider_deadline).await?; let model = read_artifact( client - .download_model(&snapshot) + .download_model(&snapshot, MODEL3D_MAX_ARTIFACT_BYTES) .await .map_err(map_provider_error)?, - ) - .await?; + )?; let preview = read_artifact( client - .download_rendered_image(&snapshot) + .download_rendered_image(&snapshot, MODEL3D_MAX_ARTIFACT_BYTES) .await .map_err(map_provider_error)?, - ) - .await?; + )?; let (preview_width, preview_height) = preview_dimensions(preview.bytes.as_slice())?; let stored_model = store_model3d_artifact( state, diff --git a/server-rs/crates/platform-tripo/examples/tripo_generation_smoke.rs b/server-rs/crates/platform-tripo/examples/tripo_generation_smoke.rs index 760233286..745d4c881 100644 --- a/server-rs/crates/platform-tripo/examples/tripo_generation_smoke.rs +++ b/server-rs/crates/platform-tripo/examples/tripo_generation_smoke.rs @@ -26,6 +26,9 @@ use tripo3d_sdk::{ClientOptions, TripoClient, WaitOptions}; const SAMPLE_IMAGE_URL: &str = "https://www.rustacean.net/assets/rustacean-flat-happy.png"; +/// 冒烟脚本单次允许的产物体积上限，与 api-server 的口径一致（512 MiB）。 +const SMOKE_MAX_ARTIFACT_BYTES: u64 = 512 * 1024 * 1024; + #[tokio::main] async fn main() -> Result<(), Box> { let api_key = required_env("TRIPO_API_KEY")?; @@ -178,26 +181,24 @@ async fn main() -> Result<(), Box> { continue; } - let downloaded = match client.download_model(&snapshot).await { - Ok(downloaded) => downloaded, + let artifact = match client + .download_model(&snapshot, SMOKE_MAX_ARTIFACT_BYTES) + .await + { + Ok(artifact) => artifact, Err(error) => { eprintln!("{name} skipped: artifact download failed: {error}"); continue; } }; - let filename = downloaded.filename(name); + let filename = artifact.filename(name); let mut file = File::create(&filename).await?; - let mut bytes = 0u64; - let mut downloaded = downloaded; - while let Some(chunk) = downloaded.next_chunk().await? { - bytes += chunk.len() as u64; - file.write_all(&chunk).await?; - } + file.write_all(&artifact.bytes).await?; println!( "{name} artifact={} bytes={} url={}", filename, - bytes, - downloaded.url.redacted() + artifact.bytes.len(), + artifact.url.redacted() ); } diff --git a/server-rs/crates/platform-tripo/src/common/client.rs b/server-rs/crates/platform-tripo/src/common/client.rs index 3bf66844f..0a7a467cd 100644 --- a/server-rs/crates/platform-tripo/src/common/client.rs +++ b/server-rs/crates/platform-tripo/src/common/client.rs @@ -4,8 +4,8 @@ use shared_contracts::model3d::common::Model3dTaskStatus; use tripo3d_sdk::TripoClient; use super::{ - TripoDownloadedArtifact, TripoError, TripoSettings, TripoTaskHandle, TripoTaskOutput, - TripoTaskSnapshot, TripoUrl, map_task, validate_task_id, + TripoArtifactBytes, TripoDownloadedArtifact, TripoError, TripoSettings, TripoTaskHandle, + TripoTaskOutput, TripoTaskSnapshot, TripoUrl, map_task, validate_task_id, }; pub struct TripoProviderClient { @@ -57,14 +57,16 @@ impl TripoProviderClient { map_task(task) } + /// 下载完整的模型产物（响应头 + 整个 body）；`max_bytes` 是单次读取的硬上限。 pub async fn download_model( &self, task: &TripoTaskSnapshot, - ) -> Result { + max_bytes: u64, + ) -> Result { let output = completed_output(task)?; // TODO SDK upstream: expose a streaming artifact API; then replace this // provider-side reqwest client with the SDK stream and remove the duplicate downloader. - self.download_artifact(&task.handle.task_id, output.model_url()) + self.download_artifact(&task.handle.task_id, output.model_url(), max_bytes) .await } @@ -75,23 +77,36 @@ impl TripoProviderClient { pub async fn download_rendered_image( &self, task: &TripoTaskSnapshot, - ) -> Result { + max_bytes: u64, + ) -> Result { let output = completed_output(task)?; - self.download_artifact(&task.handle.task_id, output.rendered_image_url()) + self.download_artifact(&task.handle.task_id, output.rendered_image_url(), max_bytes) .await } + /// 下载一个产物并读完它的 body。 + /// + /// 重试必须包住整个 body：`download_artifact_headers` 拿到响应头就返回，真正的 + /// 几十 MB 字节在 `next_chunk` 里才传输；只重试「拿到头」这一步等于没覆盖传输阶段， + /// 一次 CDN 抖动就会毁掉一次已经扣费、provider 任务也已经跑完的生成。 async fn download_artifact( &self, task_id: &str, url: &TripoUrl, - ) -> Result { + max_bytes: u64, + ) -> Result { let total_attempts = self.artifact_retries.saturating_add(1); let mut attempt = 1; loop { - match self.download_artifact_once(task_id, url).await { - Ok(downloaded) => return Ok(downloaded), + // 每次尝试都从「取响应头」开始：body 已经读到的字节随连接一起丢掉， + // 重下的是完整产物，不做断点续传（签名地址会过期，续传也只是徒增状态）。 + let result = match self.download_artifact_headers(task_id, url).await { + Ok(headers) => headers.read_all(max_bytes).await, + Err(error) => Err(error), + }; + match result { + Ok(artifact) => return Ok(artifact), Err(error) if attempt < total_attempts && error.is_retryable() => { tokio::time::sleep(download_backoff(attempt)).await; attempt += 1; @@ -101,7 +116,7 @@ impl TripoProviderClient { } } - async fn download_artifact_once( + async fn download_artifact_headers( &self, task_id: &str, url: &TripoUrl, @@ -143,6 +158,7 @@ impl TripoProviderClient { .map(str::to_owned); let content_length = response.content_length(); Ok(TripoDownloadedArtifact::new( + task_id.to_string(), url.clone(), content_type, content_length, @@ -215,15 +231,22 @@ mod tests { /// 测试用的「无数据推进」预算：够短让用例跑得快，又给 CI 抖动留出余量。 const STALL_TIMEOUT: Duration = Duration::from_millis(300); + /// 测试用产物体积上限：足够大，只有专门的超限用例才会撞上。 + const TEST_MAX_ARTIFACT_BYTES: u64 = 1024 * 1024; /// 正常推进的间隔：明显小于预算，保证用例只在「真的没数据」时才失败。 const PROGRESS_INTERVAL: Duration = Duration::from_millis(100); fn test_client() -> TripoProviderClient { + test_client_with_retries(0) + } + + /// 带产物重试次数的测试客户端：验证「整体重下」需要至少一次重试。 + fn test_client_with_retries(retries: u32) -> TripoProviderClient { let settings = TripoSettings::new( "test-key".to_string(), "http://127.0.0.1:1".to_string(), STALL_TIMEOUT, - 0, + retries, "genarrative-test-tripo/1".to_string(), ) .expect("测试配置必须合法"); @@ -314,7 +337,7 @@ mod tests { 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)) + .download_artifact("task-1", &artifact_url(addr), TEST_MAX_ARTIFACT_BYTES) .await { Ok(_) => panic!("响应头一直不来必须按无数据推进失败"), @@ -327,7 +350,7 @@ mod tests { 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)) + .download_artifact_headers("task-1", &artifact_url(addr)) .await .expect("响应头与第一块数据必须正常返回"); @@ -343,13 +366,14 @@ mod tests { .await .expect_err("传输中途断流必须按无数据推进失败"); assert!(matches!(error, TripoError::Request { .. }), "{error:?}"); + assert!(error.is_retryable(), "传输阶段失败必须可重试：{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)) + .download_artifact_headers("task-1", &artifact_url(addr)) .await .expect("响应头必须正常返回"); @@ -363,4 +387,82 @@ mod tests { } assert_eq!(received.len(), 10, "慢但持续的下载必须完整收完"); } + + /// 第一次连接在 body 中途结束：下载入口必须整体重下，第二次读完完整产物。 + #[tokio::test] + async fn artifact_download_retries_the_whole_body_after_a_broken_transfer() { + const COMPLETE_BODY: &[u8] = b"glTF\x02\x00\x00\x00complete-glb"; + 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((mut first, _)) = listener.accept().await else { + return; + }; + read_request(&mut first).await; + let head = "HTTP/1.1 200 OK\r\n\ + Content-Type: model/gltf-binary\r\n\ + Content-Length: 64\r\n\ + \r\n"; + let _ = first.write_all(head.as_bytes()).await; + let _ = first.write_all(b"glTF").await; + let _ = first.flush().await; + let _ = first.shutdown().await; + + // 第二次：完整产物。 + let Ok((mut second, _)) = listener.accept().await else { + return; + }; + read_request(&mut second).await; + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Type: model/gltf-binary\r\nContent-Length: {}\r\n\r\n", + COMPLETE_BODY.len() + ); + let _ = second.write_all(head.as_bytes()).await; + let _ = second.write_all(COMPLETE_BODY).await; + let _ = second.flush().await; + + // 第一次连接保持打开：客户端读到的是「提前结束」而不是连接被重置。 + tokio::time::sleep(Duration::from_secs(30)).await; + drop(first); + }); + + let artifact = test_client_with_retries(1) + .download_artifact("task-1", &artifact_url(addr), TEST_MAX_ARTIFACT_BYTES) + .await + .expect("body 中途失败必须整体重下并成功"); + assert_eq!(artifact.bytes, COMPLETE_BODY); + assert_eq!(artifact.content_length, Some(COMPLETE_BODY.len() as u64)); + } + + /// 产物超过调用方给的体积上限：按输出违约失败，且不重试。 + #[tokio::test] + async fn artifact_download_rejects_bodies_over_the_caller_limit() { + const BODY: &[u8] = b"glTF\x02\x00\x00\x00oversized"; + let addr = spawn_mock_server(|mut socket| async move { + read_request(&mut socket).await; + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Type: model/gltf-binary\r\nContent-Length: {}\r\n\r\n", + BODY.len() + ); + let _ = socket.write_all(head.as_bytes()).await; + let _ = socket.write_all(BODY).await; + let _ = socket.flush().await; + }) + .await; + + let error = test_client_with_retries(2) + .download_artifact("task-1", &artifact_url(addr), 8) + .await + .expect_err("超过上限的产物必须被拒绝"); + assert!( + matches!(error, TripoError::OutputSchema { .. }), + "{error:?}" + ); + assert!(!error.is_retryable(), "体积超限不是重试能解决的问题"); + } } diff --git a/server-rs/crates/platform-tripo/src/common/mod.rs b/server-rs/crates/platform-tripo/src/common/mod.rs index d6ee406c8..87a4fe4f0 100644 --- a/server-rs/crates/platform-tripo/src/common/mod.rs +++ b/server-rs/crates/platform-tripo/src/common/mod.rs @@ -12,8 +12,9 @@ pub use config::TripoSettings; pub use error::{TripoError, TripoField, TripoValidationReason}; pub(crate) use extra::extra_fields; pub(crate) use mapping::map_task; +pub(crate) use types::TripoDownloadedArtifact; pub use types::{ - TripoDownloadedArtifact, TripoTaskFailure, TripoTaskHandle, TripoTaskOutput, TripoTaskSnapshot, + TripoArtifactBytes, TripoTaskFailure, TripoTaskHandle, TripoTaskOutput, TripoTaskSnapshot, TripoTaskType, TripoUrl, }; pub(crate) use validation::{ diff --git a/server-rs/crates/platform-tripo/src/common/types.rs b/server-rs/crates/platform-tripo/src/common/types.rs index 327ac73a7..5589f04e7 100644 --- a/server-rs/crates/platform-tripo/src/common/types.rs +++ b/server-rs/crates/platform-tripo/src/common/types.rs @@ -122,25 +122,49 @@ pub struct TripoTaskSnapshot { pub completed_at: Option, } -/// provider 产物的下载句柄。模型与预览图共用同一条流式读取路径， +/// provider 产物的下载句柄（响应头阶段）。模型与预览图共用同一条流式读取路径， /// 因此类型名按“产物”而不是“模型”命名。 -pub struct TripoDownloadedArtifact { - pub url: TripoUrl, - pub content_type: Option, - pub content_length: Option, +/// +/// 只在本 crate 内使用：对外只暴露读完的 [`TripoArtifactBytes`]，流式消费等 +/// 「产物直送 OSS」落地时再决定是否公开。 +/// +/// 拿到的只是响应头：body 由 [`TripoDownloadedArtifact::next_chunk`] 逐块读出。 +/// [`crate::TripoProviderClient`] 的下载入口会把「响应头 + 读完 body」作为一个整体 +/// 重试单元（见 [`TripoDownloadedArtifact::read_all`]），所以这里的失败不等于调用方失败。 +pub(crate) struct TripoDownloadedArtifact { + task_id: String, + pub(crate) url: TripoUrl, + pub(crate) content_type: Option, + pub(crate) content_length: Option, response: Response, status: u16, received: u64, } +/// 已完整读出的 provider 产物：响应头元数据 + 全部字节。 +/// +/// 与 [`TripoDownloadedArtifact`] 的区别是「读完了」：body 中途失败会被下载入口整体重下， +/// 因此拿到这个值就代表这次读取是完整的。（几十 MB 的产物当前一次性读进内存，api-server +/// 侧同样如此；未来接 OSS 流式 / 分片上传时再回到流式句柄。） +#[derive(Debug)] +pub struct TripoArtifactBytes { + pub url: TripoUrl, + pub content_type: Option, + /// provider 声明的长度，仅用于读取时的完整性校验与内存预分配，不是落库口径。 + pub content_length: Option, + pub bytes: Vec, +} + impl TripoDownloadedArtifact { pub(crate) fn new( + task_id: String, url: TripoUrl, content_type: Option, content_length: Option, response: Response, ) -> Self { Self { + task_id, url, content_type, content_length, @@ -150,15 +174,14 @@ impl TripoDownloadedArtifact { } } - pub async fn next_chunk(&mut self) -> Result, TripoError> { + pub(crate) async fn next_chunk(&mut self) -> Result, TripoError> { let chunk = self .response .chunk() .await - .map_err(|error| TripoError::Request { - message: format!("failed to read artifact response body: {error}"), - status: Some(self.status), - })?; + // 传输错误不是 HTTP 状态：`status` 描述的是响应头，写进这里会让 + // `is_retryable` 把 200 当成「不值得重试」，CDN 抖动就再也重试不了。 + .map_err(|error| body_read_error(&self.task_id, self.status, format!("{error}")))?; match chunk { Some(chunk) => { self.received += chunk.len() as u64; @@ -167,13 +190,15 @@ impl TripoDownloadedArtifact { None => { if let Some(expected) = self.content_length { if self.received != expected { - return Err(TripoError::Request { - message: format!( + // body 提前结束同样按传输失败处理：整段重下才有机会拿到完整字节。 + return Err(body_read_error( + &self.task_id, + self.status, + format!( "artifact length mismatch: expected {expected} bytes, received {}", self.received ), - status: Some(self.status), - }); + )); } } Ok(None) @@ -181,27 +206,77 @@ impl TripoDownloadedArtifact { } } + /// 读完整个响应体。单次读取：失败不在这里重试（只有一份连接）， + /// 由下载入口用新连接整体重下。 + /// + /// `max_bytes` 是单次读取的硬上限：超过它说明响应异常或产物口径变了， + /// 宁可失败退款也不要把调用方内存打满。 + pub(crate) async fn read_all( + mut self, + max_bytes: u64, + ) -> Result { + let mut bytes = + Vec::with_capacity(self.content_length.unwrap_or(0).min(max_bytes) as usize); + while let Some(chunk) = self.next_chunk().await? { + let next_len = bytes.len() as u64 + chunk.len() as u64; + if next_len > max_bytes { + return Err(TripoError::OutputSchema { + task_id: self.task_id.clone(), + message: format!( + "artifact exceeds the {max_bytes} byte limit implied by the caller: {} bytes and still streaming", + next_len + ), + }); + } + bytes.extend_from_slice(chunk.as_ref()); + } + Ok(TripoArtifactBytes { + url: self.url, + content_type: self.content_type, + content_length: self.content_length, + bytes, + }) + } +} + +impl TripoArtifactBytes { /// 产物落盘用的文件名。远端地址不可信，扩展名只接受短的字母数字 token： /// `Url::path_segments` 已做百分号解码，`%2F` / `%5C` 这类编码分隔符会直接 /// 进到扩展名里，不过滤就会拼出跨目录的路径。 pub fn filename(&self, name: &str) -> String { - let extension = self - .url - .0 - .path_segments() - .and_then(|segments| segments.last()) - .and_then(|segment| segment.rsplit_once('.')) - .map(|(_, extension)| extension) - .filter(|extension| { - !extension.is_empty() - && extension.len() <= MAX_ARTIFACT_FILE_EXTENSION_LEN - && extension.bytes().all(|byte| byte.is_ascii_alphanumeric()) - }) - .unwrap_or_else(|| fallback_artifact_extension(self.content_type.as_deref())); - format!("{}.{extension}", sanitize_artifact_file_stem(name)) + artifact_filename(name, self.content_type.as_deref(), &self.url) } } +/// 响应体读取失败：统一成可重试的传输错误，并把 HTTP 状态留在文案里供排障。 +fn body_read_error(task_id: &str, status: u16, detail: String) -> TripoError { + TripoError::Request { + message: format!( + "failed to read artifact response body for task {task_id} (HTTP {status}): {detail}" + ), + status: None, + } +} + +/// 产物落盘用的文件名。远端地址不可信，扩展名只接受短的字母数字 token： +/// `Url::path_segments` 已做百分号解码，`%2F` / `%5C` 这类编码分隔符会直接 +/// 进到扩展名里，不过滤就会拼出跨目录的路径。 +fn artifact_filename(name: &str, content_type: Option<&str>, url: &TripoUrl) -> String { + let extension = url + .0 + .path_segments() + .and_then(|segments| segments.last()) + .and_then(|segment| segment.rsplit_once('.')) + .map(|(_, extension)| extension) + .filter(|extension| { + !extension.is_empty() + && extension.len() <= MAX_ARTIFACT_FILE_EXTENSION_LEN + && extension.bytes().all(|byte| byte.is_ascii_alphanumeric()) + }) + .unwrap_or_else(|| fallback_artifact_extension(content_type)); + format!("{}.{extension}", sanitize_artifact_file_stem(name)) +} + /// URL 路径给不出可用扩展名时按 content type 兜底：同一条下载路径也服务预览图， /// 一律写 `.glb` 会把 PNG / JPEG 产物写成模型文件名。 fn fallback_artifact_extension(content_type: Option<&str>) -> &'static str { diff --git a/server-rs/crates/platform-tripo/src/lib.rs b/server-rs/crates/platform-tripo/src/lib.rs index a409c3b36..101dfa523 100644 --- a/server-rs/crates/platform-tripo/src/lib.rs +++ b/server-rs/crates/platform-tripo/src/lib.rs @@ -9,7 +9,7 @@ pub mod multiview_to_model; pub mod text_to_model; pub use common::{ - TripoDownloadedArtifact, TripoError, TripoField, TripoProviderClient, TripoSettings, + TripoArtifactBytes, TripoError, TripoField, TripoProviderClient, TripoSettings, TripoTaskFailure, TripoTaskHandle, TripoTaskOutput, TripoTaskSnapshot, TripoTaskType, TripoUrl, TripoValidationReason, };